package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"sync"
	"testing"
	"time"

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

// restartEtag reads an app's restart-generation etag straight from the store,
// so a test asserts on what the bucket actually reports rather than a value it
// made up.
func restartEtag(t *testing.T, ctx context.Context, s release.Store, acct, app string) string {
	t.Helper()
	_, etag, err := s.Get(ctx, release.RestartGenKey(acct, app))
	if err != nil {
		t.Fatalf("read restart generation %s/%s: %v", acct, app, err)
	}
	return etag
}

// askRestart is a member's `carlos restart` landing in the bucket: the one
// write the whole feature turns on.
func askRestart(t *testing.T, ctx context.Context, s release.Store, acct, app string) {
	t.Helper()
	if err := release.TouchRestartGen(ctx, s, acct, app); err != nil {
		t.Fatalf("TouchRestartGen %s/%s: %v", acct, app, err)
	}
}

// namedUnitRoute is the addr-backed, operator-named-unit shape issue #110
// added and #113's second case is about: a console. Only the ROOT pass can
// cycle it.
func namedUnitRoute(host, acct, app, unit string) registry.Route {
	rt := execRoute(host, acct, app)
	rt.Socket = ""
	rt.Addr = "127.0.0.1:9000"
	rt.Unit = unit
	return rt
}

// recordingRestart is the root pass's dispatch seam: every host it was asked
// to cycle, in order, plus an optional per-host failure.
type recordingRestart struct {
	hosts []string
	fail  map[string]error
}

func (r *recordingRestart) fn(host string) error {
	r.hosts = append(r.hosts, host)
	if err, ok := r.fail[host]; ok {
		return err
	}
	return nil
}

// ---- the agent's sweep -------------------------------------------------------

func TestRestartGenPassFirstSightIsSilent(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven") // requested before this box ever looked
	routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
	cache := map[[2]string]string{}

	moved, errs := restartGenPass(ctx, routes, s, cache)
	if len(moved) != 0 {
		t.Fatalf("first sight moved %v — a box that has just started must not cycle everything it serves", moved)
	}
	if len(errs) != 0 {
		t.Fatalf("unexpected read errors: %v", errs)
	}
	if got, want := cache[[2]string{"bab", "eleven"}], restartEtag(t, ctx, s, "bab", "eleven"); got != want {
		t.Fatalf("cache etag = %q, want the store's %q", got, want)
	}
}

func TestRestartGenPassMovementAsksOnce(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	routes := []registry.Route{
		execRoute("a.example.com", "bab", "eleven"),
		execRoute("b.example.com", "bab", "eleven"), // same app, two routes
	}
	cache := map[[2]string]string{}
	restartGenPass(ctx, routes, s, cache) // baseline

	askRestart(t, ctx, s, "bab", "eleven")
	moved, errs := restartGenPass(ctx, routes, s, cache)
	if len(errs) != 0 {
		t.Fatalf("unexpected read errors: %v", errs)
	}
	if len(moved) != 1 || moved[0] != [2]string{"bab", "eleven"} {
		t.Fatalf("moved = %v, want exactly one app entry (one request, one ask)", moved)
	}
	if moved, _ := restartGenPass(ctx, routes, s, cache); len(moved) != 0 {
		t.Fatalf("the pass after a request moved %v — a stale cache would ask forever", moved)
	}
}

func TestRestartGenPassAbsentStaysSilentAndFirstRequestMoves(t *testing.T) {
	ctx, s := genKit(t)
	routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")}
	cache := map[[2]string]string{}

	// An app nobody has ever restarted: absent, legal, forever.
	for i := range 2 {
		moved, errs := restartGenPass(ctx, routes, s, cache)
		if len(moved) != 0 {
			t.Fatalf("pass %d: an app with no restart generation moved %v", i, moved)
		}
		if len(errs) != 0 {
			t.Fatalf("pass %d: absent is not a read error: %v", i, errs)
		}
	}
	if got := cache[[2]string{"bab", "eleven"}]; got != genAbsent {
		t.Fatalf("cache = %q, want the known-absent sentinel %q", got, genAbsent)
	}

	// Its FIRST ever request: absent -> present is movement, or the very first
	// `carlos restart` for an app would be the one that never works.
	askRestart(t, ctx, s, "bab", "eleven")
	moved, _ := restartGenPass(ctx, routes, s, cache)
	if len(moved) != 1 {
		t.Fatalf("absent -> present moved %v, want the app", moved)
	}
}

func TestRestartGenPassReadErrorLeavesCacheAndKeepsGoing(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	askRestart(t, ctx, s, "bab", "titogo")
	routes := []registry.Route{
		execRoute("a.example.com", "bab", "eleven"),
		execRoute("b.example.com", "bab", "titogo"),
	}
	cache := map[[2]string]string{}
	restartGenPass(ctx, routes, s, cache)
	before := cache[[2]string{"bab", "eleven"}]

	askRestart(t, ctx, s, "bab", "titogo") // a real request on the app that CAN be read
	blip := &failingKeyStore{Store: s, failKey: release.RestartGenKey("bab", "eleven"), err: errors.New("503 slow down")}
	moved, errs := restartGenPass(ctx, routes, blip, cache)
	if len(errs) != 1 {
		t.Fatalf("read errors = %v, want exactly one", errs)
	}
	if len(moved) != 1 || moved[0] != [2]string{"bab", "titogo"} {
		t.Fatalf("moved = %v, want only the readable app — one bad object must not stop the pass", moved)
	}
	if got := cache[[2]string{"bab", "eleven"}]; got != before {
		t.Fatalf("cache for the unreadable app = %q, want it untouched at %q (a cleared entry is a bogus first sight)", got, before)
	}
}

func TestRestartGenPassDropsVanishedApps(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	both := []registry.Route{
		execRoute("a.example.com", "bab", "eleven"),
		execRoute("b.example.com", "bab", "titogo"),
	}
	cache := map[[2]string]string{}
	restartGenPass(ctx, both, s, cache)

	restartGenPass(ctx, both[:1], s, cache) // titogo's route is gone
	if _, ok := cache[[2]string{"bab", "titogo"}]; ok {
		t.Fatal("a route that left the box kept its cache entry")
	}
	moved, _ := restartGenPass(ctx, both, s, cache) // and comes back
	if len(moved) != 0 {
		t.Fatalf("a re-added app moved %v, want a silent first sight", moved)
	}
}

// TestRestartGenPassIgnoresNonAdoptableRoutes: the membership predicate is
// release.Adoptable, the same one the adopt pass moves. A platform alias
// shares its canonical route's process, and a mid-provision route must not be
// disturbed — neither is an app this box should be reading a restart signal
// for.
func TestRestartGenPassIgnoresNonAdoptableRoutes(t *testing.T) {
	ctx, s := genKit(t)
	alias := execRoute("alias.oncarlos.com", "bab", "eleven")
	alias.Platform = true
	pending := execRoute("new.example.com", "bab", "titogo")
	pending.Status = "creating"

	cache := map[[2]string]string{}
	restartGenPass(ctx, []registry.Route{alias, pending}, s, cache)
	if len(cache) != 0 {
		t.Fatalf("cache = %v, want empty (neither route is adoptable)", cache)
	}
}

// ---- the agent's own children ------------------------------------------------

func TestBounceExecRoutesCyclesOnlyWhatRootCannot(t *testing.T) {
	moved := [][2]string{{"bab", "eleven"}}
	routes := []registry.Route{
		execRoute("exec.example.com", "bab", "eleven"),                            // the agent's child
		unitRoute("unit.example.com", "bab", "eleven"),                            // carlos-app@ — root's
		namedUnitRoute("console.example.com", "bab", "eleven", "console.service"), // #110 — root's
		execRoute("other.example.com", "bab", "titogo"),                           // a different app
	}
	var bounced []string
	n := bounceExecRoutes(routes, moved, func(h string) bool {
		bounced = append(bounced, h)
		return true
	})
	if n != 1 || !slices.Equal(bounced, []string{"exec.example.com"}) {
		t.Fatalf("bounced %v (n=%d), want only the exec-backed route of the moved app", bounced, n)
	}
}

// TestBounceExecRoutesReachesAWarmHibernator: a hibernate-flagged route is
// included on purpose. Bounce cannot start anything, so a SLEEPING instance is
// untouched — but a warm one is exactly the wedged instance `carlos restart`
// exists for, and leaving it serving would be the feature failing at its own
// headline case.
func TestBounceExecRoutesReachesAWarmHibernator(t *testing.T) {
	rt := execRoute("warm.example.com", "bab", "eleven")
	rt.Hibernate = true
	var bounced []string
	// The activator's own contract: false when not awake, true when it stopped
	// something. Here it is awake.
	bounceExecRoutes([]registry.Route{rt}, [][2]string{{"bab", "eleven"}}, func(h string) bool {
		bounced = append(bounced, h)
		return true
	})
	if !slices.Equal(bounced, []string{"warm.example.com"}) {
		t.Fatalf("bounced %v, want the warm hibernator", bounced)
	}
}

func TestBounceExecRoutesWithoutAnActivator(t *testing.T) {
	routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")}
	if n := bounceExecRoutes(routes, [][2]string{{"bab", "eleven"}}, nil); n != 0 {
		t.Fatalf("bounced %d with no activator armed, want 0 and no panic", n)
	}
}

// ---- the agent's leg, and what makes it run ----------------------------------

// TestRestartRequestLegBouncesAndRelays: one serial move's worth of work — the
// agent's own children cycled in-process, and one marker touched for the units
// only root can reach.
func TestRestartRequestLegBouncesAndRelays(t *testing.T) {
	ctx, s := genKit(t)
	dd := t.TempDir()
	marker := filepath.Join(dd, restartRequestMarker)
	routes := []registry.Route{
		execRoute("exec.example.com", "bab", "eleven"),
		unitRoute("unit.example.com", "bab", "eleven"),
	}
	var bounced []string
	bounce := func(h string) bool { bounced = append(bounced, h); return true }
	mem := newRestartMemory()

	// Baseline (what restartSeed does at arming): silent, no marker.
	askRestart(t, ctx, s, "bab", "eleven")
	if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); asked {
		t.Fatal("the baselining leg asked for a root pass — first sight must be silent")
	}
	if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("marker written on a first sight (err %v)", err)
	}

	askRestart(t, ctx, s, "bab", "eleven")
	if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); !asked {
		t.Fatal("a moved restart generation must ask for a root pass")
	}
	if !slices.Equal(bounced, []string{"exec.example.com"}) {
		t.Fatalf("bounced %v, want the exec child only (the unit route is root's)", bounced)
	}
	if _, err := os.Stat(marker); err != nil {
		t.Fatalf("no marker after a request: %v", err)
	}

	// And a leg with nothing moved neither bounces nor re-asks.
	bounced = nil
	if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); asked {
		t.Fatal("an unmoved generation asked for a root pass")
	}
	if len(bounced) != 0 {
		t.Fatalf("bounced %v with nothing moved", bounced)
	}
}

// TestFastTickRunsTheRestartLegOnASerialMove pins the wiring the redesign rests
// on: the restart check rides the SERIAL, which is what keeps an idle box at
// one GET per tick. A boot seed must not run it (that would cycle processes for
// bumps that happened while the box was down), an unchanged serial must not run
// it, and a real move must — once, after the pass set.
func TestFastTickRunsTheRestartLegOnASerialMove(t *testing.T) {
	env := newFastFixture(t)
	var passes, legs int
	d := env.deps(func() { passes++ })
	d.restartLeg = func(context.Context) {
		legs++
		if legs > passes {
			t.Errorf("the restart leg ran before/without the pass set (legs=%d passes=%d)", legs, passes)
		}
	}
	now := time.Unix(1754640000, 0)
	ctx := context.Background()

	if err := release.BumpSerial(ctx, env.store, now); err != nil {
		t.Fatalf("BumpSerial: %v", err)
	}
	fastTick(ctx, d, now) // first sight of a present serial: seed only
	if legs != 0 {
		t.Fatalf("boot seed ran the restart leg %d times, want 0", legs)
	}
	fastTick(ctx, d, now.Add(10*time.Second)) // unchanged
	if legs != 0 {
		t.Fatalf("unchanged serial ran the restart leg %d times, want 0", legs)
	}
	if err := release.BumpSerial(ctx, env.store, now); err != nil {
		t.Fatalf("BumpSerial: %v", err)
	}
	fastTick(ctx, d, now.Add(20*time.Second))
	if legs != 1 {
		t.Fatalf("a serial move ran the restart leg %d times, want 1", legs)
	}
}

// ---- the root pass -----------------------------------------------------------

func marksOf(t *testing.T, path string) *restartMarks {
	t.Helper()
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("read marks: %v", err)
	}
	m := &restartMarks{}
	if err := json.Unmarshal(raw, m); err != nil {
		t.Fatalf("decode marks %s: %v", raw, err)
	}
	return m
}

// TestRestartPassSeedsWithoutReplaying is the expensive lesson this repo has
// already paid for once: a box that has never recorded a baseline records one
// and restarts NOTHING. Every app on a freshly upgraded box has a restart
// generation the moment anyone has ever used the verb, and cycling them all on
// the first pass would take the box out.
func TestRestartPassSeedsWithoutReplaying(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	askRestart(t, ctx, s, "bab", "titogo")
	routes := []registry.Route{
		unitRoute("a.example.com", "bab", "eleven"),
		unitRoute("b.example.com", "bab", "titogo"),
	}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}

	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("the seeding pass cycled %v, want nothing", rec.hosts)
	}
	m := marksOf(t, path)
	if len(m.Apps) != 2 {
		t.Fatalf("marks = %v, want both apps baselined", m.Apps)
	}
	// And the very next pass is a quiet one, not a replay.
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("the pass after seeding cycled %v, want nothing", rec.hosts)
	}
}

// TestRestartPassCyclesOnMovement is the feature working: baseline, request,
// dispatch — for every non-hibernating route of THAT app and no other.
func TestRestartPassCyclesOnMovement(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	askRestart(t, ctx, s, "bab", "titogo")
	sleeping := unitRoute("sleep.example.com", "bab", "eleven")
	sleeping.Hibernate = true
	routes := []registry.Route{
		unitRoute("a.example.com", "bab", "eleven"),
		namedUnitRoute("console.example.com", "bab", "eleven", "console.service"),
		sleeping,
		unitRoute("z.example.com", "bab", "titogo"), // a bystander app
	}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}
	restartPass(ctx, routes, s, path, rec.fn) // baseline

	askRestart(t, ctx, s, "bab", "eleven")
	restartPass(ctx, routes, s, path, rec.fn)

	slices.Sort(rec.hosts)
	want := []string{"a.example.com", "console.example.com"}
	if !slices.Equal(rec.hosts, want) {
		t.Fatalf("cycled %v, want %v — the hibernating route must be left asleep and another app's route untouched",
			rec.hosts, want)
	}
	if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")], restartEtag(t, ctx, s, "bab", "eleven"); got != want {
		t.Fatalf("mark = %q, want the etag just acted on %q", got, want)
	}
	// Idempotent: the same request is not served twice.
	rec.hosts = nil
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("a second pass on the same generation cycled %v, want nothing", rec.hosts)
	}
}

// TestRestartPassActsOnAnAppsFirstEverRequest: an app with no object yet is
// recorded as ABSENT, and the object's later appearance is movement away from
// that recorded value. Both halves matter — without the recorded absence, the
// first `carlos restart` for every app is either dropped (if "no mark" seeds)
// or joined by a spurious cycle of every other app in the same state (if "no
// mark" dispatches).
func TestRestartPassActsOnAnAppsFirstEverRequest(t *testing.T) {
	ctx, s := genKit(t)
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}

	restartPass(ctx, routes, s, path, rec.fn) // baseline: the app has no object at all
	if got := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")]; got != restartAbsent {
		t.Fatalf("mark = %q, want the recorded absent sentinel", got)
	}

	askRestart(t, ctx, s, "bab", "eleven")
	restartPass(ctx, routes, s, path, rec.fn)
	if !slices.Equal(rec.hosts, []string{"a.example.com"}) {
		t.Fatalf("cycled %v, want the app's route on its first ever request", rec.hosts)
	}
}

// TestRestartPassSeedsAnAppItHasNeverSeen is the storm fix (review, Critical).
// A route added to this box for an app that was restarted somewhere else weeks
// ago must be RECORDED, not cycled: this box owes nothing to a request it never
// saw, and reading "no mark" as movement is what turned a registry that landed
// after the marks file into a box that try-restarted its entire route table in
// one pass.
func TestRestartPassSeedsAnAppItHasNeverSeen(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	askRestart(t, ctx, s, "bab", "titogo") // both restarted long ago, elsewhere
	first := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}
	restartPass(ctx, first, s, path, rec.fn) // this box knows about eleven only

	// titogo's route arrives now — a fresh instance, mid-provision until this
	// moment. Its app's object has existed for weeks.
	both := append(first, unitRoute("b.example.com", "bab", "titogo"))
	restartPass(ctx, both, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("a newly-seen app cycled %v, want nothing — this box never saw that request", rec.hosts)
	}
	if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "titogo")], restartEtag(t, ctx, s, "bab", "titogo"); got != want {
		t.Fatalf("mark = %q, want the seeded etag %q", got, want)
	}
	// And it is live from here: the NEXT request cycles it.
	askRestart(t, ctx, s, "bab", "titogo")
	restartPass(ctx, both, s, path, rec.fn)
	if !slices.Equal(rec.hosts, []string{"b.example.com"}) {
		t.Fatalf("cycled %v, want the newly-seeded app's route on the next real request", rec.hosts)
	}
}

// TestRestartPassWithNoRoutesWritesNothing: a box whose registry has not landed
// yet (fresh machine, /data restoring from litestream) must not be able to
// record "I know about no apps" — that file is what makes every route it later
// gains look like movement.
func TestRestartPassWithNoRoutesWritesNothing(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}

	restartPass(ctx, nil, s, path, rec.fn)
	if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a pass with zero adoptable routes wrote %s (err %v) — it knows nothing and may claim nothing", path, err)
	}
	// The registry lands; the app is seeded, not cycled.
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("cycled %v after the registry landed, want a silent baseline", rec.hosts)
	}
}

// TestRestartPassDefersABaselineWithReadErrors: a baseline that omits the apps
// a transient 503 hid is a baseline that turns those apps into movement on the
// next pass. Nothing is written until a pass can read every app.
func TestRestartPassDefersABaselineWithReadErrors(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	askRestart(t, ctx, s, "bab", "titogo")
	routes := []registry.Route{
		unitRoute("a.example.com", "bab", "eleven"),
		unitRoute("b.example.com", "bab", "titogo"),
	}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{}

	blip := &failingKeyStore{Store: s, failKey: release.RestartGenKey("bab", "eleven"), err: errors.New("503 slow down")}
	restartPass(ctx, routes, blip, path, rec.fn)
	if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a baselining pass with a read error wrote %s (err %v)", path, err)
	}

	// A clean pass baselines both, and cycles neither.
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("the clean baselining pass cycled %v, want nothing", rec.hosts)
	}
	if m := marksOf(t, path); len(m.Apps) != 2 {
		t.Fatalf("marks = %v, want both apps", m.Apps)
	}
}

// TestRestartPassRefusesWithoutALock: no lock means either a concurrent root
// pass or a /data this pass cannot write. Dispatching under the first
// double-restarts one request; under the second it restarts hourly forever,
// since the mark can never advance. Neither is allowed to happen.
func TestRestartPassRefusesWithoutALock(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	dir := t.TempDir()
	path := filepath.Join(dir, "restart-marks.json")
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	rec := &recordingRestart{}
	restartPass(ctx, routes, s, path, rec.fn) // baseline, with a working lock

	// A directory where the lock file needs to be: opening it fails (EISDIR),
	// exactly as opening it would fail on a read-only or full /data.
	if err := os.Remove(path + ".lock"); err != nil {
		t.Fatal(err)
	}
	if err := os.Mkdir(path+".lock", 0o755); err != nil {
		t.Fatal(err)
	}
	askRestart(t, ctx, s, "bab", "eleven")
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("cycled %v with no lock — an unrecordable restart repeats forever", rec.hosts)
	}
}

// TestRestartPassSerializesConcurrentPasses: three root units can run this at
// once (carlos-restart, carlos-stage, carlos-adopt). Under the lock, one
// request produces ONE dispatch however many passes race for it.
func TestRestartPassSerializesConcurrentPasses(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	var mu sync.Mutex
	var hosts []string
	dispatch := func(h string) error {
		mu.Lock()
		defer mu.Unlock()
		hosts = append(hosts, h)
		return nil
	}
	restartPass(ctx, routes, s, path, dispatch) // baseline

	askRestart(t, ctx, s, "bab", "eleven")
	var wg sync.WaitGroup
	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			restartPass(ctx, routes, s, path, dispatch)
		}()
	}
	wg.Wait()
	if len(hosts) != 1 {
		t.Fatalf("three concurrent passes dispatched %v, want exactly one — the flock is what makes a request singular", hosts)
	}
}

// TestRestartPassAdvancesTheMarkOnAFailedDispatch: a restart request is a
// point-in-time event, not a desired state. A named unit that will not come
// back must be reported, not retried on every pass until the end of time.
func TestRestartPassAdvancesTheMarkOnAFailedDispatch(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	routes := []registry.Route{namedUnitRoute("console.example.com", "bab", "eleven", "wrong.service")}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	rec := &recordingRestart{fail: map[string]error{"console.example.com": errors.New("Unit wrong.service not loaded")}}
	restartPass(ctx, routes, s, path, rec.fn) // baseline

	askRestart(t, ctx, s, "bab", "eleven")
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 1 {
		t.Fatalf("dispatched %v, want one attempt", rec.hosts)
	}
	rec.hosts = nil
	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("a failed dispatch was retried (%v) — the request must not become a standing order", rec.hosts)
	}
}

// TestRestartPassUnreadableMarksReBaseline: a wiped or corrupt marks file
// fails toward NOT restarting. The cost is one dropped request, logged and
// re-runnable; the alternative is a box that cycles everything it serves the
// moment a state file goes missing.
func TestRestartPassUnreadableMarksReBaseline(t *testing.T) {
	ctx, s := genKit(t)
	askRestart(t, ctx, s, "bab", "eleven")
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	if err := os.WriteFile(path, []byte("{not json"), 0o644); err != nil {
		t.Fatal(err)
	}
	rec := &recordingRestart{}

	restartPass(ctx, routes, s, path, rec.fn)
	if len(rec.hosts) != 0 {
		t.Fatalf("a corrupt marks file cycled %v, want nothing", rec.hosts)
	}
	if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")], restartEtag(t, ctx, s, "bab", "eleven"); got != want {
		t.Fatalf("re-baselined mark = %q, want %q", got, want)
	}
}

// ---- the handover between the two halves -------------------------------------

// TestSeedRestartMarksIsCreateOnly: the agent writes the baseline as it arms,
// so a request made before the root pass has ever run is not silently dropped
// — but it must never overwrite a mark the root pass has already advanced, or
// a served request would be served again. Absent sightings ARE recorded: that
// sentinel is what makes an app's first ever request movement.
func TestSeedRestartMarksIsCreateOnly(t *testing.T) {
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	mem := newRestartMemory()
	mem.gen = map[[2]string]string{
		{"bab", "eleven"}: "etag-1",
		{"bab", "titogo"}: genAbsent, // seen, and it has no object
	}
	mem.scope = map[[2]string]string{
		{"bab", "eleven"}: release.RestartScopeNone,
	}
	need, err := seedRestartMarks(path, mem)
	if err != nil {
		t.Fatalf("seed: %v", err)
	}
	if need {
		// The file this seed just wrote carries both memories, so there is
		// nothing for a root pass to fill in.
		t.Fatal("a seed that wrote the whole baseline still asked for a root pass")
	}
	m := marksOf(t, path)
	if m.Apps[restartMarkKey("bab", "eleven")] != "etag-1" ||
		m.Apps[restartMarkKey("bab", "titogo")] != restartAbsent {
		t.Fatalf("seeded marks = %v, want the etag and the recorded absence", m.Apps)
	}
	// The narrowed memory is seeded by the same sweep and in the same file:
	// a baseline that recorded one signal and not the other would leave the
	// root pass reading the second one as a first sight for a second time.
	if m.Scopes[restartMarkKey("bab", "eleven")] != release.RestartScopeNone {
		t.Fatalf("seeded scope marks = %v, want the recorded empty log", m.Scopes)
	}

	mem.gen[[2]string{"bab", "eleven"}] = "etag-2"
	if need, err := seedRestartMarks(path, mem); err != nil {
		t.Fatalf("second seed: %v", err)
	} else if need {
		t.Fatal("a complete baseline on disk still asked for a root pass — every agent start would ask")
	}
	if got := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")]; got != "etag-1" {
		t.Fatalf("mark = %q after a second seed, want the original %q — seeding must never overwrite", got, "etag-1")
	}
}

// TestSeedRestartMarksRefusesAnEmptyBaseline: a sweep that saw nothing is most
// often a box whose registry has not landed yet. A file claiming "this box
// knows about no apps" is exactly what makes the root pass treat every route
// the box later gains as movement.
func TestSeedRestartMarksRefusesAnEmptyBaseline(t *testing.T) {
	path := filepath.Join(t.TempDir(), "restart-marks.json")
	if need, err := seedRestartMarks(path, newRestartMemory()); err != nil {
		t.Fatalf("seed: %v", err)
	} else if need {
		t.Fatal("a sweep that saw nothing asked for a root pass; it may claim nothing at all")
	}
	if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("an empty sweep wrote %s (err %v), want no file at all", path, err)
	}
}

// TestSeededBaselineIsWhatTheRootPassTrusts closes the handover: the agent
// seeds, a member then asks, and the root pass — running for the FIRST time,
// with no memory of its own — still cycles. Without the seed that pass would
// baseline instead and the request would vanish having been answered
// "requested".
func TestSeededBaselineIsWhatTheRootPassTrusts(t *testing.T) {
	ctx, s := genKit(t)
	dd := t.TempDir()
	routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")}
	askRestart(t, ctx, s, "bab", "eleven") // an older request, already served

	// The agent's first tick: baseline in memory, baseline on disk.
	mem := newRestartMemory()
	restartGenPass(ctx, routes, s, mem.gen)
	restartScopePass(ctx, routes, s, mem.scope)
	if _, err := seedRestartMarks(restartMarksPath(dd), mem); err != nil {
		t.Fatalf("seed: %v", err)
	}

	askRestart(t, ctx, s, "bab", "eleven") // the member types `carlos restart`
	moved, _ := restartGenPass(ctx, routes, s, mem.gen)
	if len(moved) != 1 {
		t.Fatalf("the agent missed the request: moved = %v", moved)
	}
	rec := &recordingRestart{}
	restartPass(ctx, routes, s, restartMarksPath(dd), rec.fn)
	if !slices.Equal(rec.hosts, []string{"a.example.com"}) {
		t.Fatalf("the first root pass cycled %v, want the requested route", rec.hosts)
	}
}

// ---- Task 11: sidecars in the restart pass -------------------------------

// TestRestartPassCyclesSidecars: release.Adoptable requires Kind ==
// "instance" and deliberately stays that way (adopt membership, capability
// convergence, socket/backing advisories, status pages all correctly
// exclude a sidecar) — restartCandidates is the widened, NAMED opt-in
// `carlos restart -app X` needs instead.
func TestRestartPassCyclesSidecars(t *testing.T) {
	routes := []registry.Route{
		{Host: "bot.x.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bdm", Status: "ready"},
	}
	if len(restartCandidates(routes)) != 1 {
		t.Fatal("carlos restart -app X must cycle that app's sidecars too")
	}
}

// TestRestartCandidatesStillExcludesNonAdoptableNonSidecarRoutes: the
// widening is additive — everything release.Adoptable already excluded for
// a non-sidecar route (a Platform alias, a mid-provision route) stays
// excluded.
func TestRestartCandidatesStillExcludesNonAdoptableNonSidecarRoutes(t *testing.T) {
	alias := execRoute("alias.oncarlos.com", "bab", "eleven")
	alias.Platform = true
	pending := execRoute("new.example.com", "bab", "titogo")
	pending.Status = "creating"
	if got := restartCandidates([]registry.Route{alias, pending}); len(got) != 0 {
		t.Fatalf("restartCandidates = %v, want none", got)
	}
}

// TestRestartAppsIncludesASidecarOnlyApp: an app whose only route on this
// box is a sidecar must still be tracked — restartApps (via
// restartCandidates) is what both restartGenPass and restartPass use to
// decide which apps' restart generation to read at all. Without this, a
// sidecar-only app's `carlos restart` request is never even noticed.
func TestRestartAppsIncludesASidecarOnlyApp(t *testing.T) {
	routes := []registry.Route{
		{Host: "bot.x.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bdm", Status: "ready"},
	}
	apps := restartApps(routes)
	if len(apps) != 1 || apps[0] != [2]string{"bdm", "eleven"} {
		t.Fatalf("restartApps = %v, want [[bdm eleven]]", apps)
	}
}

// ---- Task 11: cycling a sidecar's own child ------------------------------

// alwaysOnSidecarRoute and hibernatingSidecarRoute are cycleWorkers'
// fixtures — Hibernate governs nothing in cycleWorkers itself (bounce's own
// awake check does the filtering), but the two shapes exist because
// activator.Service.Bounce's stop half behaves identically for both and
// cycleWorkers must too.
func sidecarRoute(host, acct, app string) registry.Route {
	return registry.Route{Host: host, Kind: "sidecar", App: app, Account: acct, Status: "ready"}
}

// TestCycleWorkersBouncesAndWakesARunningSidecar: nothing dials a worker
// the way a request dials an instance, so Bounce's stop half — reused
// verbatim from bounceExecRoutes' own bounce callback — must be paired
// with an explicit wake here, or the sidecar stays down (see
// activator.Service.Bounce's doc comment). Only a HOST BOUNCE ACTUALLY
// STOPPED gets woken.
func TestCycleWorkersBouncesAndWakesARunningSidecar(t *testing.T) {
	routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")}
	moved := [][2]string{{"bdm", "eleven"}}
	var mu sync.Mutex
	var bounced, woken []string
	bounce := func(h string) bool { mu.Lock(); defer mu.Unlock(); bounced = append(bounced, h); return true }
	wake := func(rt registry.Route) error {
		mu.Lock()
		defer mu.Unlock()
		woken = append(woken, rt.Host)
		return nil
	}
	n := cycleWorkers(routes, moved, bounce, wake)
	mu.Lock()
	bouncedSnap := slices.Clone(bounced)
	mu.Unlock()
	if n != 1 || !slices.Equal(bouncedSnap, []string{"bot.x.sidecar.internal"}) {
		t.Fatalf("bounced %v (n=%d), want exactly the sidecar", bouncedSnap, n)
	}
	// The wake is dispatched OFF cycleWorkers' return path (fix round 2) — it
	// may not have landed yet the instant the call above returns, so this
	// waits for it rather than asserting synchronously.
	waitForCond(t, "the bounced sidecar to be re-woken", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return slices.Equal(woken, []string{"bot.x.sidecar.internal"})
	})
}

// TestCycleWorkersReturnsBeforeASlowWakeCompletes is fix round 2's own
// finding, pinned directly: WakeWorker can block up to 30s per sidecar
// (the settle window), and cycleWorkers' only caller (restartRequestLeg)
// runs synchronously on the box's single per-tick convergence goroutine —
// so cycleWorkers itself must never wait on a wake, or one slow sidecar
// delays the root-restart marker touch and every other app's turn on this
// box's tick. Asserted against a deliberately slow fake wake, the same
// elapsed-time shape Task 8 used for the worker boot pass's own
// bounded-concurrency claim.
func TestCycleWorkersReturnsBeforeASlowWakeCompletes(t *testing.T) {
	routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")}
	moved := [][2]string{{"bdm", "eleven"}}
	slow := make(chan struct{})
	bounce := func(h string) bool { return true }
	wake := func(rt registry.Route) error { <-slow; return nil }

	// Run under a deadline rather than measuring elapsed time on the test's
	// own goroutine (fix round 3): under a REGRESSION cycleWorkers itself
	// never returns, so a synchronous call plus an elapsed-time check never
	// fails at all — it hangs until Go's 10-minute default test timeout,
	// which is exactly what forced fix round 2's mutation check to run
	// under an external `timeout` to see anything fail. Wrapping the call
	// in its own goroutine and racing it against time.After (the same
	// shape TestDispatchSidecarWakesBoundsConcurrency already uses) makes a
	// regression here fail cleanly and promptly, in this test alone.
	var n int
	done := make(chan struct{})
	go func() {
		n = cycleWorkers(routes, moved, bounce, wake)
		close(done)
	}()
	select {
	case <-done:
		close(slow) // release the parked wake goroutine so the test can exit cleanly
	case <-time.After(time.Second):
		close(slow)
		t.Fatal("cycleWorkers did not return within 1s with a wake blocked indefinitely — the wake must be dispatched asynchronously, not awaited")
	}

	if n != 1 {
		t.Fatalf("cycleWorkers = %d, want 1 (the bounce itself is synchronous)", n)
	}
}

// TestCycleWorkersWakesEveryBouncedSidecarEventually: dispatching wakes
// asynchronously must not mean dropping any of them — every sidecar
// cycleWorkers bounces still gets woken, just not before this function
// returns.
func TestCycleWorkersWakesEveryBouncedSidecarEventually(t *testing.T) {
	routes := []registry.Route{
		sidecarRoute("bot.a.sidecar.internal", "bdm", "eleven"),
		sidecarRoute("bot.b.sidecar.internal", "bdm", "eleven"),
		sidecarRoute("bot.c.sidecar.internal", "bdm", "eleven"),
	}
	moved := [][2]string{{"bdm", "eleven"}}
	var mu sync.Mutex
	woken := map[string]bool{}
	bounce := func(h string) bool { return true }
	wake := func(rt registry.Route) error {
		mu.Lock()
		defer mu.Unlock()
		woken[rt.Host] = true
		return nil
	}
	if n := cycleWorkers(routes, moved, bounce, wake); n != 3 {
		t.Fatalf("cycleWorkers = %d, want 3", n)
	}
	waitForCond(t, "all three sidecars to be re-woken", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(woken) == 3
	})
}

// TestCycleWorkersFailingWakeDoesNotStopTheOthers: a wake error is logged
// (not asserted here — bounceExecRoutes' own analogous case doesn't pin the
// log line either) and must never prevent another sidecar's wake from
// running, nor cycleWorkers' own return.
func TestCycleWorkersFailingWakeDoesNotStopTheOthers(t *testing.T) {
	routes := []registry.Route{
		sidecarRoute("bot.bad.sidecar.internal", "bdm", "eleven"),
		sidecarRoute("bot.good.sidecar.internal", "bdm", "eleven"),
	}
	moved := [][2]string{{"bdm", "eleven"}}
	var mu sync.Mutex
	woken := map[string]bool{}
	bounce := func(h string) bool { return true }
	wake := func(rt registry.Route) error {
		if rt.Host == "bot.bad.sidecar.internal" {
			return errors.New("boom")
		}
		mu.Lock()
		defer mu.Unlock()
		woken[rt.Host] = true
		return nil
	}
	if n := cycleWorkers(routes, moved, bounce, wake); n != 2 {
		t.Fatalf("cycleWorkers = %d, want 2 — both were bounced regardless of the wake outcome", n)
	}
	waitForCond(t, "the good sidecar to be re-woken despite the other's wake failing", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return woken["bot.good.sidecar.internal"]
	})
}

// TestCycleWorkersLeavesASleepingSidecarAsleep: Bounce is a no-op on
// anything not awake, and cycleWorkers must never wake a sleeper just
// because a restart was requested — the same "restarting a process that is
// not running is meaningless" rule restartPass applies to a hibernating
// instance.
func TestCycleWorkersLeavesASleepingSidecarAsleep(t *testing.T) {
	routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")}
	moved := [][2]string{{"bdm", "eleven"}}
	var woken []string
	bounce := func(h string) bool { return false } // asleep: Bounce's own contract
	wake := func(rt registry.Route) error { woken = append(woken, rt.Host); return nil }
	if n := cycleWorkers(routes, moved, bounce, wake); n != 0 {
		t.Fatalf("cycleWorkers = %d, want 0 — nothing was running", n)
	}
	if len(woken) != 0 {
		t.Fatalf("woken %v, want none — a sleeping sidecar must stay asleep", woken)
	}
}

// TestCycleWorkersOnlyTouchesMovedAppsAndSidecarRoutes: the same
// app-scoping bounceExecRoutes gives an instance, and never an ordinary
// instance route even for a moved app (that is bounceExecRoutes' job, not
// this one's).
func TestCycleWorkersOnlyTouchesMovedAppsAndSidecarRoutes(t *testing.T) {
	routes := []registry.Route{
		sidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven"),
		sidecarRoute("bot.titogo.sidecar.internal", "bdm", "titogo"), // different app
		execRoute("inst.example.com", "bdm", "eleven"),               // not a sidecar
	}
	moved := [][2]string{{"bdm", "eleven"}}
	var bounced []string
	bounce := func(h string) bool { bounced = append(bounced, h); return true }
	wake := func(rt registry.Route) error { return nil }
	if n := cycleWorkers(routes, moved, bounce, wake); n != 1 || !slices.Equal(bounced, []string{"bot.eleven.sidecar.internal"}) {
		t.Fatalf("bounced %v (n=%d), want only the moved app's sidecar", bounced, n)
	}
}

// TestCycleWorkersWithoutAnActivatorOrWakeIsANoop: a box with no activator
// armed gets nil for both callbacks — a working no-op, not a panic.
func TestCycleWorkersWithoutAnActivatorOrWakeIsANoop(t *testing.T) {
	routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")}
	if n := cycleWorkers(routes, [][2]string{{"bdm", "eleven"}}, nil, nil); n != 0 {
		t.Fatalf("cycleWorkers = %d with no activator armed, want 0 and no panic", n)
	}
}

// TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued is review round
// 2's own regression pin: a NARROW earlier version of the fan-out
// (fix round 1) acquired the workerBootConcurrency semaphore in cycleWorkers'
// OWN loop, one route at a time — so once workerBootConcurrency (8) wakes
// were already in flight, the (workerBootConcurrency+1)th route's
// acquisition blocked cycleWorkers itself, and therefore restartRequestLeg
// and fastTick's single per-box convergence goroutine, for up to
// wakeWaitBound (30s) per sidecar past the 8th — a narrow reintroduction of
// exactly the bug TestCycleWorkersReturnsBeforeASlowWakeCompletes already
// pins for a SINGLE sidecar. This drives strictly more than
// workerBootConcurrency sidecars, every wake deliberately parked, and
// asserts cycleWorkers still returns immediately: the shape an app like
// Eleven's (a sidecar per enrolled member) hits routinely, not on rare
// unlucky timing.
func TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued(t *testing.T) {
	n := workerBootConcurrency + 5 // strictly more than the bound
	routes := make([]registry.Route, n)
	for i := range routes {
		routes[i] = sidecarRoute(fmt.Sprintf("s%d.x.sidecar.internal", i), "bdm", "eleven")
	}
	moved := [][2]string{{"bdm", "eleven"}}
	bounce := func(h string) bool { return true }
	slow := make(chan struct{})
	var mu sync.Mutex
	woken := map[string]bool{}
	wake := func(rt registry.Route) error {
		<-slow
		mu.Lock()
		woken[rt.Host] = true
		mu.Unlock()
		return nil
	}

	// Same deadline-vs-elapsed-time correction as
	// TestCycleWorkersReturnsBeforeASlowWakeCompletes (fix round 3): a
	// synchronous call on the test's own goroutine never fails under a
	// regression, it hangs — this is the test fix round 2's own mutation
	// check needed an external `timeout` to see fail at all.
	var got int
	done := make(chan struct{})
	go func() {
		got = cycleWorkers(routes, moved, bounce, wake)
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(time.Second):
		close(slow)
		t.Fatalf("cycleWorkers did not return within 1s with %d wakes blocked (bound %d) — the whole bounded fan-out must run off the caller's goroutine, not just each individual wake", n, workerBootConcurrency)
	}
	if got != n {
		t.Fatalf("cycleWorkers = %d, want %d (every route bounced)", got, n)
	}

	close(slow)
	waitForCond(t, "every queued sidecar to be woken eventually", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(woken) == n
	})
}