package activator

// Ported from lchat activator_test.go.
//
// Unit tests for the activator state machine (activator.go) with a fake
// backend — wake coalescing, idle→sleep, lease-lost stand-down, contention
// 503s. No AWS, no processes. The router-level tests use a small in-package
// test router that mimics the edge's dispatch (internal/edge) — deliberately
// NOT imported: the activator satisfies edge.Activator structurally, and the
// dependency must point edge→activator only.

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"time"

	"golang.org/x/sys/unix"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/seeds"

	_ "modernc.org/sqlite"
)

// --- fakes -------------------------------------------------------------------

type fakeRep struct {
	mu     sync.Mutex
	err    error
	closed bool
	closes int
	// b, when set, is bumped on Close. production's Rep.Close "does the
	// final checkpoint/sync and releases the lease" (Rep's doc comment), so
	// a successful Close is the fake's equivalent of the lease coming free
	// once a wake has gotten past Replicate (before that point, the raw
	// Lease.Release calls below count it instead: the two are mutually
	// exclusive per wake, since Replicate spends the Lease).
	b *fakeBackend
	// onClose, when set, runs at the start of Close: the moment the lease is
	// given back. Task 12 checks there that no generation still runs.
	onClose func()
}

func (r *fakeRep) Err() error {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.err
}
func (r *fakeRep) setErr(err error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.err = err
}
func (r *fakeRep) Close(context.Context) error {
	r.mu.Lock()
	hook := r.onClose
	r.mu.Unlock()
	if hook != nil {
		hook()
	}
	r.mu.Lock()
	r.closed = true
	r.closes++ // Close IS the lease release: exactly once, on every path
	b := r.b
	r.mu.Unlock()
	if b != nil {
		b.mu.Lock()
		b.releases++
		b.mu.Unlock()
	}
	return nil
}

func (r *fakeRep) setOnClose(fn func()) {
	r.mu.Lock()
	r.onClose = fn
	r.mu.Unlock()
}

func (r *fakeRep) closeCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.closes
}
func (r *fakeRep) isClosed() bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.closed
}

type fakeProc struct {
	// launchEnv is what LaunchEnv answers (merge ruling M3); nil answers
	// an empty environment, as a generation launched with no files has.
	launchEnv map[string]string
	mu        sync.Mutex
	stopped   bool
	dead      bool
	// stopGate, when non-nil, blocks Stop() until closed — holds an entry in
	// the stopping state so a test can poke at it.
	stopGate chan struct{}
	// crashed marks a NON-ZERO exit — the fact that tells a hibernating
	// worker's crash from its completion.
	crashed bool
	// stuck models a Proc whose termination cannot be proven within Stop's
	// bounded ctx (design §6.3, unitProc's ErrStillRunning): while stuck and
	// not yet dead, Stop reports ErrStillRunning instead of succeeding; exit()
	// clears it. stopAttempts counts every Stop() call, proven or not, so a
	// test can see the sweeper retrying a stop it hasn't been able to finish.
	stuck        bool
	stopAttempts int
	// stopEnters counts Stop calls ENTERED, before the gate: a test waits on
	// it to know a gated Stop is in flight.
	stopEnters int
	// label + rec feed the backend's order log ("instance" / "sidecar"), so
	// the co-spawn/co-stop tests can assert interleavings.
	label string
	rec   *fakeBackend
	// polls counts Exited() calls; exitAt, when > 0, makes the process exit
	// on that call: a death a test places inside an observation (a settle
	// window) by the observation's own polling, with no sleep.
	polls  int
	exitAt int
	// onStopEnter, when set, runs as Stop is entered, before the gate: a
	// test records what was already true when the stop began.
	onStopEnter func()
}

func (p *fakeProc) LaunchEnv() map[string]string {
	if p.launchEnv == nil {
		return map[string]string{}
	}
	return p.launchEnv
}

func (p *fakeProc) Stop(context.Context) error {
	p.mu.Lock()
	p.stopEnters++
	gate, enter := p.stopGate, p.onStopEnter
	p.mu.Unlock()
	if enter != nil {
		enter()
	}
	if gate != nil {
		<-gate
	}
	p.mu.Lock()
	p.stopAttempts++
	if p.stuck && !p.dead {
		p.mu.Unlock()
		return ErrStillRunning
	}
	p.stopped = true
	// A stopped process HAS exited: modelling that is what lets a test see the
	// window a converge respawn opens — old child down, new one not yet
	// installed — which the sweeper must not mistake for a crash.
	p.dead = true
	p.mu.Unlock()
	if p.rec != nil {
		p.rec.record("stop:" + p.label)
	}
	return nil
}

// setStopGate installs (or clears) the gate Stop blocks on, under the proc's
// own lock — the sweeper may be reading it concurrently.
func (p *fakeProc) setStopGate(gate chan struct{}) {
	p.mu.Lock()
	p.stopGate = gate
	p.mu.Unlock()
}
func (p *fakeProc) Exited() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.polls++
	if p.exitAt > 0 && p.polls >= p.exitAt {
		p.dead = true
	}
	return p.dead
}

// exitOnPoll makes the process exit on its n-th Exited() call from now.
func (p *fakeProc) exitOnPoll(n int) {
	p.mu.Lock()
	p.exitAt = p.polls + n
	p.mu.Unlock()
}

// pollCount is how many times Exited() has been asked.
func (p *fakeProc) pollCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.polls
}

// Crashed models the exec child's own answer (backend_exec.go's childProc):
// only a definitely non-zero exit is a crash. kill() therefore models a CLEAN
// exit — a bot that finished its work — and crash() models a child that fell
// over.
func (p *fakeProc) Crashed() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.dead && p.crashed
}

// crash models the child dying with a NON-ZERO status.
func (p *fakeProc) crash() {
	p.mu.Lock()
	p.dead, p.crashed = true, true
	p.mu.Unlock()
}
func (p *fakeProc) isStopped() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.stopped
}
func (p *fakeProc) stopEntered() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.stopEnters
}
func (p *fakeProc) stopCalls() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.stopAttempts
}

// exit marks the process as having died on its own (a crash, or, for a
// stuck proc, the delayed termination Stop couldn't prove at the time).
func (p *fakeProc) exit() {
	p.mu.Lock()
	p.dead = true
	p.mu.Unlock()
}

type fakeLease struct {
	rep *fakeRep
	b   *fakeBackend // records the sync interval the wake threads through
	// owed: a Release failed and RetryRelease has not yet landed. Under b.mu.
	owed bool
}

func (l *fakeLease) Replicate(_ context.Context, _ string, syncInterval time.Duration) (Rep, error) {
	if l.b != nil {
		l.b.mu.Lock()
		l.b.syncIntervals = append(l.b.syncIntervals, syncInterval)
		l.b.mu.Unlock()
	}
	return l.rep, nil
}

// Err is the held lease's liveness (leaseHealth): the renewal the wake's
// replica will inherit. A test loses the lease before Replicate by setting
// the error on the rep this lease hands out (b.lastRep().setErr).
func (l *fakeLease) Err() error {
	if l.rep == nil {
		return nil
	}
	return l.rep.Err()
}

// Release counts on the backend: every doWake bail-out must give the lease
// back exactly once, and the only way to assert that from outside is to count
// the calls the wake made. It also records its context's error: the real
// release is an S3 DeleteObject(ctx), which fails at once on a dead context
// and spends the handle anyway.
func (l *fakeLease) Release(ctx context.Context) error {
	if l.b != nil {
		l.b.mu.Lock()
		defer l.b.mu.Unlock()
		l.b.releaseCtxErrs = append(l.b.releaseCtxErrs, ctx.Err())
		if len(l.b.releaseErrs) > 0 {
			err := l.b.releaseErrs[0]
			l.b.releaseErrs = l.b.releaseErrs[1:]
			l.owed = true
			return err
		}
		l.b.releases++
		l.b.queuedAtRelease = append(l.b.queuedAtRelease, l.b.queuedJobs)
	}
	return nil
}

// RetryRelease is lease.Held's: it discharges what a failed Release left
// owed, and counts as the release once it lands.
func (l *fakeLease) RetryRelease(context.Context) error {
	if l.b == nil {
		return nil
	}
	l.b.mu.Lock()
	defer l.b.mu.Unlock()
	l.b.retries++
	if !l.owed {
		return nil
	}
	if len(l.b.retryErrs) > 0 {
		err := l.b.retryErrs[0]
		l.b.retryErrs = l.b.retryErrs[1:]
		if l.b.leaseLapsed {
			l.owed = false
			return fmt.Errorf("%w: %w", lease.ErrReleaseLapsed, err)
		}
		return err
	}
	l.owed = false
	l.b.releases++
	l.b.queuedAtRelease = append(l.b.queuedAtRelease, l.b.queuedJobs)
	return nil
}

// LegacyRunGone is ExecBackend's: the legacy run unit's stop, which
// replaces a queued start job, and its proof.
func (b *fakeBackend) LegacyRunGone(ctx context.Context, _ registry.Route) error {
	b.mu.Lock()
	b.legacyStops++
	block := b.legacyBlock
	b.mu.Unlock()
	if block {
		<-ctx.Done()
		b.mu.Lock()
		b.legacyEnded++
		b.mu.Unlock()
		return ctx.Err()
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.legacyStopErr != nil {
		return b.legacyStopErr
	}
	b.queuedJobs = 0
	return nil
}

type fakeBackend struct {
	mu         sync.Mutex
	acquires   int
	admits     int // AdmitConfig calls
	restores   int
	starts     int
	seeds      int
	releases   int // lease releases — the failure paths' assertion
	acquireErr error
	// releaseCtxErrs is ctx.Err() at each fakeLease.Release, in order.
	releaseCtxErrs []error
	// admitErr is what AdmitConfig answers: the config gate a retained-lease
	// respawn must ask again (amadan #213).
	admitErr error
	// admitHook, when set, runs at the start of AdmitConfig, off b.mu: a
	// test's window inside a respawn's preparation, after its first
	// ownership check and before its exec.
	admitHook func()
	// restoreFresh is what Restore reports: a first boot (local DB absent AND
	// no replica). False by default, so every pre-seed test doubles as the
	// "a restored instance is never seeded" case.
	restoreFresh bool
	restoreErr   error
	seedErr      error
	seedRoutes   []registry.Route
	startErr     error
	startGate    chan struct{} // when non-nil, Start() blocks until closed
	// startQueued, when non-nil, blocks Start() until it is closed or the
	// wake's context ends. It models a nested legacy start whose systemctl
	// start job systemd has queued behind an edge restart (#292). The
	// instance-control client gives up when its context does, but the
	// queued job SURVIVES that: a cancelled start counts one in queuedJobs
	// and answers as ExecBackend does (no process, legacyStartUncertain).
	// Only LegacyRunGone, the unit's stop, clears it.
	startQueued chan struct{}
	queuedJobs  int
	// legacyStops counts LegacyRunGone calls; legacyStopErr is what they
	// answer (an unproven stop) while set.
	legacyStops   int
	legacyStopErr error
	// legacyBlock holds every LegacyRunGone until its context ends (a
	// stop that never proves); legacyEnded counts those that returned.
	legacyBlock bool
	legacyEnded int
	// startLegacyFail fails every Start as a legacy run unit's start root
	// may still carry out, at once, outside any shutdown.
	startLegacyFail bool
	// leaseLapsed: the owed records have passed their expiry, so a retry
	// that fails again answers lease.ErrReleaseLapsed, as lease.Held does.
	leaseLapsed bool
	// queuedAtRelease is queuedJobs at each lease release that landed: a
	// release over a still-queued job is the single-writer violation.
	queuedAtRelease []int
	// releaseErrs fails the next Release calls, in order (a store that
	// refuses or times out the DELETE); retryErrs does the same for
	// RetryRelease. retries counts RetryRelease calls.
	releaseErrs []error
	retryErrs   []error
	retries     int
	reps        []*fakeRep
	procs       []*fakeProc
	// syncIntervals records, per wake, the replication-class tick doWake
	// threads into Replicate() — the ladder's seam assertion.
	syncIntervals []time.Duration
	dueVal        int64 // what the instance reports at the sleep-time next-due probe
	dueErr        error // when non-nil, the probe fails (dead socket, missing token, …)
	// dueProbeCount counts NextDue calls. A worker has no socket to answer on,
	// so the assertion its sleep needs is "the probe never ran at all", not
	// "the probe returned nothing".
	dueProbeCount int

	// execVerFn models the live/<host> target the child execs at spawn — Start()
	// returns it as the spawnVersion, exactly as ExecBackend reads the symlink
	// just before Start(). Nil → "" (unknown), which disables convergence, so
	// tests that don't care are unaffected. Point it at the same source as the
	// sweep's liveVer seam to model a symlink faithfully (both read the current
	// target).
	execVerFn func() string

	// Sidecar co-spawn. sidecarStaged is HasSidecar's answer (false by
	// default, so every pre-co-spawn test doubles as the "host without a
	// sidecar spawns none" case); sidecarErr injects spawn failures.
	sidecarStaged bool
	sidecarErr    error
	sidecarStarts int
	sidecars      []*fakeProc
	// Worker spawn (StartWorker) — mirrors the sidecar fields above for the
	// same reason: workerStaged... no, a worker has no HasSidecar-style probe
	// (Task 6's territory); workerErr injects spawn failures, workerEnvs
	// records what env each call was handed so a caller-side test can assert
	// the per-wake token actually reached the backend.
	workerErr    error
	workers      []*fakeProc
	workerEnvs   []map[string]string
	workerStartN int
	// workerDeadOnStart hands back a child that is ALREADY dead — the sidecar
	// that crashes on boot, which the settle window exists to catch.
	workerDeadOnStart bool
	workerSpawnVer    string // what StartWorker reports as the build it execed
	// workerStartDelay holds StartWorker open for a beat, so a test can land a
	// sweep INSIDE a converge respawn's window (old child stopped, new one not
	// yet installed) rather than hoping to hit it.
	workerStartDelay time.Duration
	workerEnterN     int // StartWorker calls ENTERED (before the delay)
	// workerStopGate, when set, is installed on every worker child spawned
	// from then on: its Stop blocks until the gate closes, so a test can hold
	// a child mid-drain and watch what happens around it.
	workerStopGate chan struct{}
	// workerStuck makes every worker child spawned from then on a stuck
	// fakeProc: its Stop answers ErrStillRunning until exit().
	workerStuck bool
	// workerStartHook, when set, runs INSIDE StartWorker (off b.mu), so a test
	// can model what the child itself does while its wake is still resolving —
	// notably posting its next due time over the agent socket during the
	// settle window.
	workerStartHook func(rt registry.Route)
	// sidecarGate, when non-nil, blocks StartSidecar() until closed. Models
	// a stalled restart call so a test can assert it doesn't hold a.mu.
	sidecarGate chan struct{}
	// sidecarStartEntered, when non-nil, is closed the first time
	// StartSidecar is entered (before it waits on sidecarGate). Lets a test
	// wait for a restart's call to actually begin rather than sleeping a
	// guessed duration.
	sidecarStartEntered     chan struct{}
	sidecarStartEnteredOnce sync.Once
	// sidecarLabel, when set, labels the sidecars StartSidecar returns from
	// then on ("sidecar" otherwise), so the order log can tell them apart.
	sidecarLabel string
	// sidecarStopGate and sidecarStuck, when set, are given to the sidecars
	// StartSidecar returns from then on (fakeProc's stopGate and stuck): a
	// returned sidecar whose stop blocks, or cannot prove its exit.
	sidecarStopGate chan struct{}
	sidecarStuck    bool
	// order interleaves backend calls and proc stops (start:instance,
	// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
	// assertions.
	order []string

	// procStuck makes every Start() return a fakeProc whose Stop reports
	// ErrStillRunning until the test calls its exit() (design §6.3).
	procStuck bool
	// startedRoutes records the registry.Route Start() actually received,
	// per call. doWake's route-refresh assertion (design §6.8) needs to see
	// what the backend was actually handed, not what the caller passed in.
	startedRoutes []registry.Route
	// layoutFn, when set, is what Layout reports: a test's way to exercise
	// doWake's pre-Acquire fence check without a real ExecBackend. Nil means
	// LayoutFlat, "": every test that never sets this field gets the
	// behavior it always had, since a flat layout never fences a wake.
	layoutFn func(rt registry.Route) (Layout, string)

	// ident is the identity capability (settings spec §4.2, Task 5c),
	// opted into per test (wakeidentity_test.go). Guarded by mu.
	ident identityFake
}

// Layout is fakeBackend's Classifier implementation, always present so
// doWake's type assertion succeeds; layoutFn (above) decides what it says.
func (b *fakeBackend) Layout(rt registry.Route) (Layout, string) {
	if b.layoutFn == nil {
		return LayoutFlat, ""
	}
	return b.layoutFn(rt)
}

func (b *fakeBackend) record(ev string) {
	b.mu.Lock()
	b.order = append(b.order, ev)
	b.mu.Unlock()
}

func (b *fakeBackend) events() []string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return append([]string(nil), b.order...)
}

func (b *fakeBackend) Acquire(context.Context, registry.Route) (Lease, error) {
	b.mu.Lock()
	b.acquires++
	err := b.acquireErr
	rep := &fakeRep{b: b}
	b.reps = append(b.reps, rep)
	b.mu.Unlock()
	if err != nil {
		return nil, err
	}
	return &fakeLease{rep: rep, b: b}, nil
}

func (b *fakeBackend) AdmitConfig(context.Context, registry.Route) error {
	b.mu.Lock()
	hook := b.admitHook
	b.mu.Unlock()
	if hook != nil {
		hook() // a respawn's preparation, held open by the test
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.admits++
	if b.admitErr != nil {
		return &errConfigRefused{err: b.admitErr}
	}
	return nil
}

func (b *fakeBackend) Restore(context.Context, registry.Route) (bool, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.restores++
	b.order = append(b.order, "restore")
	return b.restoreFresh, b.restoreErr
}

func (b *fakeBackend) Seed(_ context.Context, rt registry.Route) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.seeds++
	b.order = append(b.order, "seed")
	b.seedRoutes = append(b.seedRoutes, rt)
	return b.seedErr
}

func (b *fakeBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
	b.mu.Lock()
	b.starts++
	b.order = append(b.order, "start:instance")
	gate := b.startGate
	queued := b.startQueued
	fn := b.execVerFn
	startErr := b.startErr
	proc := &fakeProc{label: "instance", rec: b, stuck: b.procStuck}
	b.procs = append(b.procs, proc)
	b.startedRoutes = append(b.startedRoutes, rt)
	b.mu.Unlock()
	if startErr != nil {
		return nil, "", startErr
	}
	b.mu.Lock()
	legacyFail := b.startLegacyFail
	b.mu.Unlock()
	if legacyFail {
		return nil, "", markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: root did not answer", rt.Host)))
	}
	if gate != nil {
		<-gate
	}
	if queued != nil {
		select {
		case <-queued:
		case <-ctx.Done():
			b.mu.Lock()
			b.queuedJobs++
			b.mu.Unlock()
			return nil, "", markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, ctx.Err())))
		}
	}
	ver := ""
	if fn != nil {
		ver = fn()
	}
	return proc, ver, nil
}

func (b *fakeBackend) NextDue(context.Context, registry.Route) (int64, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.dueProbeCount++
	b.order = append(b.order, "nextDue")
	if b.dueErr != nil {
		return 0, b.dueErr
	}
	return b.dueVal, nil
}

func (b *fakeBackend) HasSidecar(registry.Route) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.sidecarStaged
}

func (b *fakeBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
	b.mu.Lock()
	gate := b.sidecarGate
	entered := b.sidecarStartEntered
	b.mu.Unlock()
	if entered != nil {
		b.sidecarStartEnteredOnce.Do(func() { close(entered) })
	}
	if gate != nil {
		<-gate
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.sidecarStarts++
	b.order = append(b.order, "start:sidecar")
	if b.sidecarErr != nil {
		return nil, b.sidecarErr
	}
	label := "sidecar"
	if b.sidecarLabel != "" {
		label = b.sidecarLabel
	}
	p := &fakeProc{label: label, rec: b, stopGate: b.sidecarStopGate, stuck: b.sidecarStuck}
	b.sidecars = append(b.sidecars, p)
	return p, nil
}

func (b *fakeBackend) sidecarStartCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.sidecarStarts
}

func (b *fakeBackend) lastSidecar() *fakeProc {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.sidecars[len(b.sidecars)-1]
}

func (b *fakeBackend) setSidecarErr(err error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.sidecarErr = err
}

func (b *fakeBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) {
	b.mu.Lock()
	hook := b.workerStartHook
	delay := b.workerStartDelay
	b.workerEnterN++ // counted BEFORE the delay: "the exec has begun"
	b.mu.Unlock()
	if delay > 0 {
		// A slow exec — ExecBackend materializes the build from the bucket
		// when live/<host> dangles — and, like the real one, abandonable: a
		// cancelled context produces NO child, which is what makes a shutdown
		// that cancels distinguishable from one that merely outran the spawn.
		select {
		case <-time.After(delay):
		case <-ctx.Done():
			return nil, "", ctx.Err()
		}
	}
	if hook != nil {
		hook(rt) // the child, acting while its wake is still in flight
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.workerStartN++
	b.order = append(b.order, "startWorker:"+rt.Host)
	b.workerEnvs = append(b.workerEnvs, env)
	if b.workerErr != nil {
		return nil, "", b.workerErr
	}
	p := &fakeProc{label: "worker", rec: b, dead: b.workerDeadOnStart, stopGate: b.workerStopGate, stuck: b.workerStuck}
	b.workers = append(b.workers, p)
	// Same symlink model as Start: execVerFn, when wired, is the live/<host>
	// target this exec would have read — so a convergence test can move the
	// symlink and have the respawned child report the new build.
	ver := b.workerSpawnVer
	if b.execVerFn != nil {
		ver = b.execVerFn()
	}
	return p, ver, nil
}

// lastWorker is the most recently spawned worker child.
func (b *fakeBackend) lastWorker() *fakeProc {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.workers) == 0 {
		return nil
	}
	return b.workers[len(b.workers)-1]
}

// kill models the child dying on its own (a crash), which the sweep notices
// through Exited().
func (p *fakeProc) kill() {
	p.mu.Lock()
	p.dead = true
	p.mu.Unlock()
}

// workerStarts counts StartWorker calls; startCount counts Start (the socket
// path, with its health probe) calls. A worker must only ever appear in the
// first.
func (b *fakeBackend) workerStarts() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.workerStartN
}

// workerEnters counts spawns that have BEGUN — the only way to know a test is
// inside a slow exec rather than merely near one.
func (b *fakeBackend) workerEnters() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.workerEnterN
}

func (b *fakeBackend) startCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.starts
}

func (b *fakeBackend) dueProbes() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.dueProbeCount
}

func (b *fakeBackend) lastWorkerEnv() map[string]string {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.workerEnvs) == 0 {
		return nil
	}
	return b.workerEnvs[len(b.workerEnvs)-1]
}

func (b *fakeBackend) setDue(v int64) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.dueVal = v
}

func (b *fakeBackend) counts() (acquires, restores, starts int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.acquires, b.restores, b.starts
}

func (b *fakeBackend) seedCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.seeds
}

func (b *fakeBackend) releaseCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.releases
}

// releaseCtxErrors is ctx.Err() as each fakeLease.Release saw it.
func (b *fakeBackend) releaseCtxErrors() []error {
	b.mu.Lock()
	defer b.mu.Unlock()
	return slices.Clone(b.releaseCtxErrs)
}

func (b *fakeBackend) lastProc() *fakeProc {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.procs[len(b.procs)-1]
}

func (b *fakeBackend) lastStartedRoute() registry.Route {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.startedRoutes[len(b.startedRoutes)-1]
}

func (b *fakeBackend) lastRep() *fakeRep {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.reps[len(b.reps)-1]
}

// fakeClock is the activator's `now` seam.
type fakeClock struct {
	mu sync.Mutex
	t  time.Time
}

func (c *fakeClock) now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.t
}
func (c *fakeClock) advance(d time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.t = c.t.Add(d)
}

// The registry IS the DueStore in production — assert the wiring compiles.
var _ DueStore = (*registry.Registry)(nil)

// testActivator wires the fakes to a REAL (temp-dir) registry as the due store,
// so the due-time tests exercise the actual routes.next_due SQL, not a fake of it.
func testActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeClock, *registry.Registry) {
	t.Helper()
	b := &fakeBackend{}
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	a := New(Config{Bucket: "test-bucket", Idle: idle}, b, reg)
	a.now = clk.now
	return a, b, clk, reg
}

func testRoute(host string) registry.Route {
	// Hibernate is set because these routes model activator-managed hosts —
	// the per-host gate (the due sweep's flag filter, ServeHTTP's gate) only
	// acts on flagged instances.
	return registry.Route{Host: host, Kind: "instance", Socket: "/tmp/nope.sock",
		DBPath: filepath.Join("/tmp", host+".db"), Hibernate: true}
}

// testWorkerRoute is a standalone sidecar's row: no socket (nothing can dial
// it), an argv of its own, and the hibernate flag that makes the sweeper's
// idle clock apply. Status "ready" mirrors the due-sweep's gate.
func testWorkerRoute(host string) registry.Route {
	return registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`,
		DBPath: filepath.Join("/tmp", host+".db"), Hibernate: true, Status: "ready"}
}

// contains: membership in the backend's order log.
func contains(events []string, ev string) bool { return slices.Contains(events, ev) }

// newTestServiceAwake builds a Service against the fake backend and drives
// host to actAwake via a real beginRequest/release round trip — the same
// wake arrange block used throughout this file (e.g.
// TestActivatorShutdownStopsChildren, TestActivatorConvergesOnSymlinkRepoint).
func newTestServiceAwake(t *testing.T, host string) *Service {
	t.Helper()
	a, _, _, _ := testActivator(t, time.Hour)
	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest %s: %v", host, err)
	}
	release()
	return a
}

// arrangeAwake drives rt's host to actAwake on an already-built Service via a
// real beginRequest/release round trip — the same arrange block
// newTestServiceAwake uses, but against a caller-supplied Service/route
// instead of building its own.
func arrangeAwake(t *testing.T, a *Service, rt registry.Route) {
	t.Helper()
	release, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatalf("beginRequest %s: %v", rt.Host, err)
	}
	release()
}

// okHandler is a proxy stand-in that writes 200 OK — for tests asserting
// only that a request reached the handler, not what a real instance would
// return.
func okHandler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
	})
}

// armTimer puts one armed schedule row on host, due at next.
func armTimer(t *testing.T, reg *registry.Registry, host string, next time.Time) {
	t.Helper()
	if err := reg.UpsertTimer(registry.Timer{
		Host: host, Name: "sweep", Kind: "schedule", Spec: "every 5m",
		Path: "/jobs/sweep", NextAt: next.Unix(), DueAt: next.Unix(),
	}); err != nil {
		t.Fatalf("UpsertTimer %s: %v", host, err)
	}
}

// waitFor polls until a condition holds — the stop path runs on its own
// goroutine, so tests observe its completion rather than assuming it.
func waitFor(t *testing.T, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %s", what)
}

// sidecarSettled reports whether host has no sidecar restart in flight. The
// call runs off a.mu (design §6.7), so a test driving several restarts back
// to back (each expecting the previous one's count and install to have
// already landed) waits on this rather than racing sweepOnce's own, now much
// shorter, return.
func (a *Service) sidecarSettled(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	return e == nil || !e.sidecarStarting
}

func (a *Service) entryCount() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	return len(a.entries)
}

// --- the test router ---------------------------------------------------------
//
// A minimal stand-in for internal/edge's dispatch: routes with a socket get a
// unix reverse proxy; hibernate-flagged instance routes go through the
// activator's ServeHTTP hook first (the same shape as edge.Edge.ServeHTTP);
// reload() re-lists the registry and calls RetireExcept with the managed set,
// exactly as edge.Reload does.

type testUpstream struct {
	proxy http.Handler
	route registry.Route
	hib   bool
}

type testRouter struct {
	reg *registry.Registry
	act *Service
	mu  sync.Mutex
	up  map[string]testUpstream
}

func newTestRouter(reg *registry.Registry, act *Service) (*testRouter, error) {
	rt := &testRouter{reg: reg, act: act}
	if err := rt.reload(); err != nil {
		return nil, err
	}
	return rt, nil
}

func (rt *testRouter) reload() error {
	routes, err := rt.reg.List()
	if err != nil {
		return err
	}
	m := make(map[string]testUpstream)
	managed := make(map[string]bool)
	for _, t := range routes {
		hib := rt.act != nil && t.Kind == "instance" && t.Hibernate
		if hib {
			managed[t.Host] = true
		}
		// Socket wins over Addr, mirroring edge.Reload's upstream choice.
		proxy := testUnixProxy(t.Socket)
		if t.Socket == "" && t.Addr != "" {
			proxy = testTCPProxy(t.Addr)
		}
		m[t.Host] = testUpstream{proxy: proxy, route: t, hib: hib}
	}
	rt.mu.Lock()
	rt.up = m
	rt.mu.Unlock()
	if rt.act != nil {
		rt.act.RetireExcept(managed)
	}
	return nil
}

func (rt *testRouter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	host := r.Host
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}
	rt.mu.Lock()
	up, ok := rt.up[host]
	rt.mu.Unlock()
	if !ok {
		http.Error(w, "unknown host", http.StatusNotFound)
		return
	}
	if up.hib && rt.act.ServeHTTP(w, r, up.route, up.proxy) {
		return
	}
	up.proxy.ServeHTTP(w, r)
}

func testUnixProxy(socket string) http.Handler {
	dialer := &net.Dialer{}
	return &httputil.ReverseProxy{
		Rewrite: func(pr *httputil.ProxyRequest) {
			pr.Out.URL.Scheme = "http"
			pr.Out.URL.Host = "carlos" // placeholder; the unix dialer ignores it
			pr.Out.Host = pr.In.Host
		},
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				return dialer.DialContext(ctx, "unix", socket)
			},
		},
	}
}

// testTCPProxy mirrors edge's tcpProxy: the upstream shape for whole-app
// tenants on a TCP Addr (the unix-socket fallback).
func testTCPProxy(addr string) http.Handler {
	return &httputil.ReverseProxy{
		Rewrite: func(pr *httputil.ProxyRequest) {
			pr.Out.URL.Scheme = "http"
			pr.Out.URL.Host = addr
			pr.Out.Host = pr.In.Host
		},
	}
}

// proxyGet sends GET base+"/" with an overridden Host header and returns the
// response body + status code.
func proxyGet(t *testing.T, base, host string) (string, int) {
	t.Helper()
	req, err := http.NewRequest(http.MethodGet, base+"/", nil)
	if err != nil {
		t.Fatalf("new request: %v", err)
	}
	req.Host = host
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return string(b), resp.StatusCode
}

// requireUnixSockets skips a test when the sandbox forbids unix-socket
// creation (probe: net.Listen("unix")), matching internal/host's helper. Most
// tests here fake the backend over TCP; only the ones that bind a real unix
// listener need this.
func requireUnixSockets(t *testing.T) {
	t.Helper()
	l, err := net.Listen("unix", tmpSock(t))
	if err != nil {
		if strings.Contains(err.Error(), "operation not permitted") {
			t.Skipf("sandbox forbids unix sockets: %v", err)
		}
		t.Fatalf("unix socket probe: %v", err)
	}
	l.Close()
}

// tmpSockSeq numbers tmpSock's paths, so tests running in parallel never
// share one.
var tmpSockSeq atomic.Uint64

// tmpSock returns a short unix socket path (the sun_path limit is ~104 bytes,
// so t.TempDir() can be too long) and schedules its removal.
func tmpSock(t *testing.T) string {
	t.Helper()
	s := filepath.Join(os.TempDir(), fmt.Sprintf("carlos-act-%d-%d.sock", os.Getpid(), tmpSockSeq.Add(1)))
	os.Remove(s)
	t.Cleanup(func() { os.Remove(s) })
	return s
}

// --- tests ---------------------------------------------------------------------

// Many concurrent requests during a wake coalesce into exactly one wake: one
// lease claim, one restore, one spawn.
func TestActivatorWakeCoalesces(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Minute)
	b.startGate = make(chan struct{})

	const n = 12
	var ok atomic.Int32
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
			if err != nil {
				t.Errorf("beginRequest: %v", err)
				return
			}
			ok.Add(1)
			release()
		}()
	}
	// Let the requests pile up on the gated wake, then open it.
	waitFor(t, "the wake to begin", func() bool { _, _, s := b.counts(); return s == 1 })
	time.Sleep(50 * time.Millisecond)
	close(b.startGate)
	wg.Wait()

	if got := ok.Load(); got != n {
		t.Fatalf("%d/%d requests succeeded", got, n)
	}
	acquires, restores, starts := b.counts()
	if acquires != 1 || restores != 1 || starts != 1 {
		t.Fatalf("wake did not coalesce: acquires=%d restores=%d starts=%d (want 1 each)", acquires, restores, starts)
	}
}

// The request that starts a wake is marked as the one that woke the app
// (edge.MarkWoke, read back by the edge's request records); a request that
// queues behind the same wake is not, though it waited just as long.
func TestActivatorMarksOnlyTheWakingRequestWoke(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Minute)
	b.startGate = make(chan struct{})
	rt := testRoute("woke.example.com")
	proxy := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })

	serve := func(req *http.Request) <-chan int {
		done := make(chan int, 1)
		go func() {
			out := httptest.NewRecorder()
			if !a.ServeHTTP(out, req, rt, proxy) {
				t.Error("activator did not handle a hibernating route")
			}
			done <- out.Code
		}()
		return done
	}
	first, _ := edge.WithAnswerFlag(httptest.NewRequest("GET", "https://woke.example.com/", nil))
	firstDone := serve(first)
	// The first request holds the gated wake before the second arrives, so
	// the second can only join it.
	waitFor(t, "the wake to begin", func() bool { _, _, s := b.counts(); return s == 1 })
	second, _ := edge.WithAnswerFlag(httptest.NewRequest("GET", "https://woke.example.com/", nil))
	secondDone := serve(second)
	time.Sleep(50 * time.Millisecond)
	close(b.startGate)
	if c1, c2 := <-firstDone, <-secondDone; c1 != http.StatusOK || c2 != http.StatusOK {
		t.Fatalf("statuses %d, %d; want 200, 200", c1, c2)
	}
	if !edge.Woke(first) {
		t.Fatal("the request that started the wake is not marked woke")
	}
	if edge.Woke(second) {
		t.Fatal("a request that joined the wake is marked woke")
	}
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("starts = %d, want one wake", starts)
	}
}

// OnWake (customer-fleets plan, Task 6: a fleet box's binding publish) must
// fire after a successful wake, named with the right host, and — the whole
// point of running it via `go a.OnWake(host)` rather than calling it inline —
// must NOT be able to hold the wake open: beginRequest returns and releases
// long before the hook is allowed to finish.
func TestActivatorWakeCallsOnWakeHookAsync(t *testing.T) {
	a, _, _, _ := testActivator(t, time.Minute)
	block := make(chan struct{})
	hookHost := make(chan string, 1)
	a.OnWake = func(rt registry.Route) {
		<-block // would hang beginRequest forever if invoked synchronously
		hookHost <- rt.Host
	}

	done := make(chan struct{})
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute("hook.example.com"))
		if err != nil {
			t.Errorf("beginRequest: %v", err)
			return
		}
		release()
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("beginRequest never returned — OnWake appears to be blocking the wake path")
	}

	close(block)
	select {
	case host := <-hookHost:
		if host != "hook.example.com" {
			t.Fatalf("OnWake called with %q, want hook.example.com", host)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("OnWake was never called after a successful wake")
	}
}

// A nil OnWake (every non-fleet deployment, and every test above this one)
// must be a pure no-op — no crash, wake behaves exactly as before Task 6.
func TestActivatorNilOnWakeIsNoop(t *testing.T) {
	a, _, _, _ := testActivator(t, time.Minute)
	release, err := a.beginRequest(context.Background(), testRoute("nohook.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
}

// The wake threads the route's replication class into Replicate() as the
// litestream sync interval — the per-plan durability ladder's seam. Unset
// (every pre-ladder route) must thread 0, i.e. litestream's default: a deploy
// of the ladder changes no host's durability by itself.
func TestActivatorWakeThreadsReplicationClass(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)

	for i, tc := range []struct {
		class string
		want  time.Duration
	}{
		{"", 0},
		{replicationStream, 0},
		{replicationBatched, 5 * time.Minute},
		{replicationDaily, 24 * time.Hour},
	} {
		rt := testRoute(fmt.Sprintf("ladder%d.example.com", i))
		rt.Replication = tc.class
		release, err := a.beginRequest(context.Background(), rt)
		if err != nil {
			t.Fatalf("beginRequest(%q): %v", tc.class, err)
		}
		release()
		b.mu.Lock()
		got := b.syncIntervals[len(b.syncIntervals)-1]
		b.mu.Unlock()
		if got != tc.want {
			t.Errorf("class %q threaded sync interval %v, want %v", tc.class, got, tc.want)
		}
	}
}

// --- first-wake seed ---------------------------------------------------------
//
// The seam spec §6 pins: fresh means the local DB was absent AND no replica
// existed — a first boot, never a restore — and the route's seed runs strictly
// between Restore and Start, i.e. before the first health probe ever reaches
// the instance. A brand-new instance is therefore born seeded, and a restored
// one is never re-seeded over its own data.

func TestActivatorSeedsFreshInstanceBeforeStart(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreFresh = true
	rt := testRoute("fresh.example.com")
	rt.Seed = `["seed"]`

	release, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	if got := b.seedCount(); got != 1 {
		t.Fatalf("seed calls = %d, want 1 on a fresh boot with a seed", got)
	}
	ev := b.events()
	r, s, st := indexOf(ev, "restore"), indexOf(ev, "seed"), indexOf(ev, "start:instance")
	if r == -1 || s == -1 || st == -1 || !(r < s && s < st) {
		t.Fatalf("seed order wrong: %v (want restore < seed < start:instance — seeded before the first health probe)", ev)
	}
	b.mu.Lock()
	seededHost := b.seedRoutes[0].Host
	b.mu.Unlock()
	if seededHost != rt.Host {
		t.Fatalf("seeded host = %q, want %q", seededHost, rt.Host)
	}
}

// A restore is never a first boot: the instance's data came back from the
// bucket, and running the seed over it is exactly the corruption this fresh
// flag exists to prevent.
func TestActivatorSkipsSeedWhenRestored(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreFresh = false
	rt := testRoute("restored.example.com")
	rt.Seed = `["seed"]`

	release, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	if got := b.seedCount(); got != 0 {
		t.Fatalf("seed calls = %d on a restored instance, want 0", got)
	}
}

// No seed declared: a fresh boot just boots (every route added before this
// feature, and every app that creates its own schema).
func TestActivatorSkipsSeedWhenRouteHasNone(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreFresh = true
	rt := testRoute("noseed.example.com") // rt.Seed == ""

	release, err := a.beginRequest(context.Background(), rt)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	if got := b.seedCount(); got != 0 {
		t.Fatalf("seed calls = %d for a route with no seed, want 0", got)
	}
}

// A failed seed fails the WHOLE wake — a half-seeded instance must not serve —
// and gives the lease back exactly once, the Start-failure precedent. Nothing
// is spawned, and the entry is gone so the next request retries from scratch.
func TestActivatorSeedFailureFailsWakeAndReleasesLease(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreFresh = true
	b.seedErr = errors.New("seed exited 1")
	rt := testRoute("badseed.example.com")
	rt.Seed = `["seed"]`

	_, err := a.beginRequest(context.Background(), rt)
	if err == nil {
		t.Fatal("wake succeeded despite a failing seed")
	}
	if !strings.Contains(err.Error(), "seed "+rt.Host) {
		t.Fatalf("wake error = %q, want it to name the seed and the host", err)
	}
	if _, _, starts := b.counts(); starts != 0 {
		t.Fatalf("instance was spawned after a failed seed (starts=%d)", starts)
	}
	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases = %d after a failed seed, want exactly 1", got)
	}
	if a.entryCount() != 0 {
		t.Fatal("failed wake left an entry behind")
	}
}

// TestActivatorSeedUnprovenLeavesTheLeaseToItsTTL pins design §6.2: a seed
// whose cgroup never emptied may still have a SIGKILL survivor writing the
// database, so doWake must not release the lease over it the way it does
// for every other seed failure (design §6.3's ErrStillRunning gets the same
// treatment in stop()).
func TestActivatorSeedUnprovenLeavesTheLeaseToItsTTL(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreFresh = true
	b.seedErr = fmt.Errorf("seed for badseed.example.com: %w: processes remain after a successful seed: still running", ErrSeedUnproven)
	rt := testRoute("badseed.example.com")
	rt.Seed = `["seed"]`

	_, err := a.beginRequest(context.Background(), rt)
	if err == nil {
		t.Fatal("wake succeeded despite an unproven seed")
	}
	if !errors.Is(err, ErrSeedUnproven) {
		t.Fatalf("wake error = %v, want it to wrap ErrSeedUnproven", err)
	}
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d for an unproven seed, want exactly 0", got)
	}
}

// Task 8, plan 3: a wake whose Restore refuses because an earlier seed's
// processes are still not proven gone (the <db>.seed-unproven obligation)
// treats it as the seed's own ErrSeedUnproven: nothing is seeded or
// started, and the lease is left to its TTL.
func TestActivatorRestoreRefusedForAnUnprovenSeedLeavesTheLeaseToItsTTL(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreErr = fmt.Errorf("unproven.example.com: %w: an earlier seed's processes are not proven gone", ErrSeedUnproven)
	rt := testRoute("unproven.example.com")
	rt.Seed = `["seed"]`

	_, err := a.beginRequest(context.Background(), rt)
	if !errors.Is(err, ErrSeedUnproven) {
		t.Fatalf("wake error = %v, want it to wrap ErrSeedUnproven", err)
	}
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d, want 0: something of the earlier seed may still write the database", got)
	}
	b.mu.Lock()
	seeds, starts := b.seeds, b.starts
	b.mu.Unlock()
	if seeds != 0 || starts != 0 {
		t.Fatalf("seeds=%d starts=%d after a refused restore, want neither", seeds, starts)
	}
}

// execSeedBackend is fakeBackend with the real ExecBackend's Restore and
// Seed, so a test can drive doWake through the seed obligation on disk.
type execSeedBackend struct {
	*fakeBackend
	eb *ExecBackend
}

func (x *execSeedBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
	x.record("restore")
	return x.eb.Restore(ctx, rt)
}

func (x *execSeedBackend) Seed(ctx context.Context, rt registry.Route) error {
	x.record("seed")
	return x.eb.Seed(ctx, rt)
}

// Task 8 fix round 1, finding 8: the next wake after an unproven seed,
// driven through doWake. The first wake's seed fails and its processes are
// not proven gone: the lease stays and the obligation is on disk. The next
// wake, while the proof still fails, is refused before any seed or start,
// with the lease kept and the database untouched. Once the proof holds, the
// wake removes the database, seeds again and starts.
func TestTheNextWakeAfterAnUnprovenSeedWaitsForItsProof(t *testing.T) {
	const host = "next-wake.example.com"
	a, b, _, _ := testActivator(t, time.Minute)
	eb, _ := delegatedBackend(t, host)
	a.backend = &execSeedBackend{fakeBackend: b, eb: eb}
	runs := filepath.Join(t.TempDir(), "runs")
	stageSeedScript(t, eb, host, "v1", fmt.Sprintf(
		"echo run >> %q\nprintf half > \"$3\"\n[ \"$(wc -l < %q)\" -ge 2 ] && exit 0\nexit 1\n", runs, runs))
	swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
		return false, nil
	})
	rt := testRoute(host)
	rt.DBPath = ""
	rt.Seed = `["seed"]`
	db := filepath.Join(eb.DataDir, host+".db")
	seedRuns := func() int {
		raw, _ := os.ReadFile(runs)
		return strings.Count(string(raw), "run\n")
	}

	if _, err := a.beginRequest(context.Background(), rt); !errors.Is(err, ErrSeedUnproven) {
		t.Fatalf("first wake = %v, want ErrSeedUnproven", err)
	}
	if _, err := os.Stat(db + seedUnprovenSuffix); err != nil {
		t.Fatalf("no obligation after the first wake: %v", err)
	}

	// The proof still fails (a temp dir is no cgroupfs): refused.
	if _, err := a.beginRequest(context.Background(), rt); !errors.Is(err, ErrSeedUnproven) {
		t.Fatalf("second wake = %v, want ErrSeedUnproven", err)
	}
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d, want 0 while the seed is unproven", got)
	}
	if n, starts := seedRuns(), b.startCount(); n != 1 || starts != 0 {
		t.Fatalf("seed runs = %d, starts = %d after a refused wake; want 1 and 0", n, starts)
	}
	if got, _ := os.ReadFile(db); string(got) != "half" {
		t.Fatalf("database = %q after a refused wake, want it untouched", got)
	}

	// The proof holds: the wake reseeds and starts.
	eb.cgroupKill = func(context.Context, registry.Route, string) error { return nil }
	if _, err := a.beginRequest(context.Background(), rt); err != nil {
		t.Fatalf("third wake = %v, want it to reseed and start", err)
	}
	if n, starts := seedRuns(), b.startCount(); n != 2 || starts != 1 {
		t.Fatalf("seed runs = %d, starts = %d; want the seed to have run again and the instance started", n, starts)
	}
	if _, err := os.Stat(db + seedUnprovenSuffix); !os.IsNotExist(err) {
		t.Fatalf("obligation still there (%v) after a proven wake", err)
	}
}

// TestActivatorWakeRefusesFencedRowBeforeAcquire pins design §6.8: a fenced
// row is refused at the top of doWake, before the lease is ever acquired,
// not only inside Start after the lease is already held.
func TestActivatorWakeRefusesFencedRowBeforeAcquire(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	const reason = "layout migration in flight"
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFenced, reason }
	rt := testRoute("fenced.example.com")

	_, err := a.beginRequest(context.Background(), rt)
	if err == nil {
		t.Fatal("wake succeeded against a fenced row")
	}
	if !strings.Contains(err.Error(), reason) {
		t.Fatalf("wake error = %q, want it to carry the fence reason %q", err, reason)
	}
	if acquires, _, starts := b.counts(); acquires != 0 || starts != 0 {
		t.Fatalf("fenced row reached the backend (acquires=%d starts=%d), want neither", acquires, starts)
	}
}

// The other two doWake bail-outs release the lease exactly once too — the
// property the seed step slots into the middle of.
func TestActivatorWakeFailuresReleaseLeaseOnce(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name   string
		inject func(*fakeBackend)
		want   string
	}{
		{"restore", func(b *fakeBackend) { b.restoreErr = errors.New("s3 down") }, "restore "},
		{"start", func(b *fakeBackend) { b.startErr = errors.New("never healthy") }, "start "},
	} {
		t.Run(tc.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Minute)
			tc.inject(b)
			rt := testRoute(tc.name + "-fail.example.com")

			_, err := a.beginRequest(context.Background(), rt)
			if err == nil {
				t.Fatalf("wake succeeded despite a failing %s", tc.name)
			}
			if !strings.Contains(err.Error(), tc.want+rt.Host) {
				t.Fatalf("wake error = %q, want it to name %q and the host", err, tc.want)
			}
			if got := b.releaseCount(); got != 1 {
				t.Fatalf("lease releases = %d after a failed %s, want exactly 1", got, tc.name)
			}
			if a.entryCount() != 0 {
				t.Fatal("failed wake left an entry behind")
			}
		})
	}
}

// The singleflight hole behind flake #184: the wake INITIATOR must wait on
// the entry it created, never re-read the map — a fast-FAILING wake deletes
// its entry, and an initiator that looks the host up again finds nothing and
// starts a second wake: two lease acquires, two restore attempts, doubled
// error events, for one request. The window is one preemption wide (between
// the initiator's unlock and its next lock), so this drives many failed
// wakes under -race and contending goroutines and asserts the invariant
// acquires == requests exactly.
func TestActivatorFailedWakeIsSingleflight(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreErr = errors.New("s3 down")
	rt := testRoute("singleflight.example.com")

	// Background lock churn widens the initiator's unlock→lock window the
	// way a loaded test suite does.
	stopChurn := make(chan struct{})
	var churn sync.WaitGroup
	for i := 0; i < 4; i++ {
		churn.Add(1)
		go func() {
			defer churn.Done()
			for {
				select {
				case <-stopChurn:
					return
				default:
					a.mu.Lock()
					_ = len(a.entries) //nolint:staticcheck // hold briefly
					a.mu.Unlock()
				}
			}
		}()
	}

	const requests = 300
	for i := 0; i < requests; i++ {
		if _, err := a.beginRequest(context.Background(), rt); err == nil {
			t.Fatal("wake succeeded despite a failing restore")
		}
	}
	close(stopChurn)
	churn.Wait()

	b.mu.Lock()
	acquires, releases := b.acquires, b.releases
	b.mu.Unlock()
	if acquires != requests || releases != requests {
		t.Fatalf("acquires=%d releases=%d for %d requests — a failed wake was not singleflight (#184)",
			acquires, releases, requests)
	}
	if a.entryCount() != 0 {
		t.Fatal("failed wakes left entries behind")
	}
}

// A restore failure never reaches the seed: seeding a host whose data we
// couldn't materialize is precisely the "half-seeded instance" case.
func TestActivatorRestoreFailureSkipsSeed(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.restoreErr = errors.New("s3 down")
	rt := testRoute("restorefail.example.com")
	rt.Seed = `["seed"]`

	if _, err := a.beginRequest(context.Background(), rt); err == nil {
		t.Fatal("wake succeeded despite a failing restore")
	}
	if got := b.seedCount(); got != 0 {
		t.Fatalf("seed calls = %d after a failed restore, want 0", got)
	}
}

// After the idle window with nothing in flight, the sweeper hibernates the
// instance: process stopped, replication closed (final checkpoint + lease
// release), entry forgotten; the next request wakes it afresh.
func TestActivatorIdleSleep(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	// Not yet idle: nothing happens.
	clk.advance(9 * time.Second)
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("instance hibernated before its idle window elapsed")
	}

	clk.advance(2 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("idle instance's process was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("idle instance's replication was not closed (no final checkpoint/lease release)")
	}

	// A new request wakes it again — a second full wake.
	release, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest after sleep: %v", err)
	}
	release()
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d after sleep+rewake, want 2", starts)
	}
}

// An in-flight request (an open WebSocket, in production) holds the instance
// awake past the idle window; only after it finishes does the idle clock run.
func TestActivatorInflightKeepsAwake(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}

	clk.advance(time.Hour)
	a.sweepOnce()
	if a.entryCount() != 1 || b.lastProc().isStopped() {
		t.Fatal("instance with an in-flight request was hibernated")
	}

	release() // stamps lastActive = now
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("instance hibernated immediately after its last request finished")
	}
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
}

// A ship repoints live/<host> under an awake child. A running process never
// re-reads its symlink, so idle-hibernate is the only thing that converges an
// instance — and an always-awake host (every sidecar host) never idles.
// The sweep closes that gap: it notices the drift and, after a grace window and
// rate-limited to one per tick, gracefully cycles the child so its next wake
// execs the new binary.
func TestActivatorConvergesOnSymlinkRepoint(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour) // idle far away — isolate the convergence path
	var ver atomic.Value                        // the live/<host> symlink target; swapped to model a ship
	ver.Store("v1")
	// Model one symlink: the child execs whatever it points at now (execVerFn,
	// frozen into spawnVersion at wake), and the sweep reads the current target
	// (liveVer). A "ship" swaps ver; the running child keeps its frozen version.
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	// Symlink unchanged: never cycles, however long it stays awake.
	clk.advance(3 * ConvergeGrace)
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("cycled an instance whose symlink never moved")
	}

	// Ship: repoint the symlink. The first sweep after only starts the grace
	// clock — it must not cycle yet, so a move-then-rollback never drops a live
	// connection.
	ver.Store("v2")
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("cycled inside the grace window — a transient repoint would drop a live connection")
	}

	// Past the grace: gracefully cycle (stop → final checkpoint → lease release →
	// asleep). The reconnecting client/sidecar would wake it onto v2.
	clk.advance(ConvergeGrace + time.Second)
	a.sweepOnce()
	waitFor(t, "convergence cycle", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("stale instance's process was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("convergence cycle skipped the final checkpoint / lease release")
	}

	// Re-wake against the now-current symlink: spawnVersion tracks v2, so it
	// never cycles again — no thrash.
	release, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("re-wake after convergence: %v", err)
	}
	release()
	clk.advance(3 * ConvergeGrace)
	a.sweepOnce()
	if a.entryCount() != 1 {
		t.Fatal("cycled again after already converging onto the current build")
	}
}

// Two awake children go stale at once (a fleet-wide ship). One sweep cycles at
// most one — a staggered convergence, never a simultaneous blip of every space.
func TestActivatorConvergenceRateLimitsPerSweep(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	var ver atomic.Value
	ver.Store("v1")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }

	for _, h := range []string{"one.example.com", "two.example.com"} {
		release, err := a.beginRequest(context.Background(), testRoute(h))
		if err != nil {
			t.Fatalf("beginRequest %s: %v", h, err)
		}
		release()
	}
	if a.entryCount() != 2 {
		t.Fatalf("want 2 awake instances, got %d", a.entryCount())
	}

	ver.Store("v2") // ship under both
	a.sweepOnce()   // starts the grace clock for both
	clk.advance(ConvergeGrace + time.Second)

	a.sweepOnce() // both are past grace, but only one may cycle this tick
	waitFor(t, "first convergence cycle", func() bool { return a.entryCount() == 1 })
	// A moment later, the sweep takes the second.
	a.sweepOnce()
	waitFor(t, "second convergence cycle", func() bool { return a.entryCount() == 0 })
}

// The feature's whole premise: converge a host that NEVER reaches inflight==0
// (an open WebSocket, or a sidecar's persistent connection), which
// idle-hibernate can never touch. Unlike idle-hibernate, convergence fires with
// a request in flight — the graceful stop drains it exactly as a restart would.
func TestActivatorConvergesWithRequestInFlight(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	var ver atomic.Value
	ver.Store("v1")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }

	// Hold the request open for the whole test — inflight stays 1, so the idle
	// path (which needs inflight==0) can never converge this host.
	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	defer release()

	ver.Store("v2") // ship under the busy host
	a.sweepOnce()   // grace starts; must not cycle a busy host mid-grace
	if a.entryCount() != 1 {
		t.Fatal("cycled inside the grace window")
	}
	clk.advance(ConvergeGrace + time.Second)
	a.sweepOnce()
	// Cycles despite the in-flight request — idle-hibernate never would.
	waitFor(t, "busy-host convergence", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("busy stale instance's process was not stopped")
	}
}

// ConvergeDecision's state machine, exhaustively: the grace clock starts on
// first drift, survives a rollback correctly, and a read error never clears
// accrued grace (it means "unknown", not "converged").
func TestConvergeDecision(t *testing.T) {
	base := time.Unix(1_700_000_000, 0)
	set := base.Add(-40 * time.Second) // a grace clock that started 40s ago
	errBoom := errors.New("readlink boom")
	cases := []struct {
		name       string
		cur        string
		readErr    error
		spawnVer   string
		staleSince time.Time
		wantCycle  bool
		wantStale  time.Time
	}{
		{"matches — clears grace", "v1", nil, "v1", set, false, time.Time{}},
		{"empty cur — no data, clears", "", nil, "v1", set, false, time.Time{}},
		{"empty spawnVer — unknown, clears", "v2", nil, "", set, false, time.Time{}},
		{"first drift — starts grace, no cycle", "v2", nil, "v1", time.Time{}, false, base},
		{"drift within grace — holds, no cycle", "v2", nil, "v1", base.Add(-30 * time.Second), false, base.Add(-30 * time.Second)},
		{"drift past grace — cycles", "v2", nil, "v1", base.Add(-ConvergeGrace - time.Second), true, base.Add(-ConvergeGrace - time.Second)},
		{"read error — preserves accrued grace", "", errBoom, "v1", set, false, set},
		{"read error before any drift — stays zero", "", errBoom, "v1", time.Time{}, false, time.Time{}},
		{"rollback to spawn — clears grace", "v1", nil, "v1", base.Add(-ConvergeGrace - time.Second), false, time.Time{}},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			cycle, next := ConvergeDecision(c.cur, c.readErr, c.spawnVer, c.staleSince, base, ConvergeGrace)
			if cycle != c.wantCycle {
				t.Errorf("cycle = %v, want %v", cycle, c.wantCycle)
			}
			if !next.Equal(c.wantStale) {
				t.Errorf("nextStale = %v, want %v", next, c.wantStale)
			}
		})
	}
}

// A lost lease (failed renewal) stands the instance down immediately — no
// waiting for idle: someone else may own the S3 prefix now.
func TestActivatorLeaseLostStandsDown(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	b.lastRep().setErr(errors.New("lease renewal failed"))
	a.sweepOnce() // idle window is an hour away; the lost lease alone must trigger the stop
	waitFor(t, "stand-down", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("lease-lost instance's process was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("lease-lost instance's replication was not closed")
	}
}

// A dead child (crashed instance) is recycled by the next request: the stale
// entry is stopped and a fresh wake happens in the same beginRequest call.
func TestActivatorDeadChildRewakesOnRequest(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	p := b.lastProc()
	p.mu.Lock()
	p.dead = true
	p.mu.Unlock()

	release, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest after crash: %v", err)
	}
	release()
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d after crash+request, want 2 (recycle + rewake)", starts)
	}
}

// A request that arrives mid-stop waits for the stop, then wakes fresh.
func TestActivatorRequestDuringStopRewakes(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)

	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	// Hold the stop open so the entry sits in stopping.
	gate := make(chan struct{})
	p := b.lastProc()
	p.mu.Lock()
	p.stopGate = gate
	p.mu.Unlock()

	clk.advance(time.Minute)
	a.sweepOnce()

	got := make(chan error, 1)
	go func() {
		release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
		if err == nil {
			release()
		}
		got <- err
	}()
	time.Sleep(50 * time.Millisecond) // the request should now be parked on e.stopped
	close(gate)
	if err := <-got; err != nil {
		t.Fatalf("request during stop: %v", err)
	}
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d, want 2 (sleep then rewake)", starts)
	}
}

// A wake that exceeds the bound turns queued requests away with errWakeTimeout
// (503 at the edge) rather than piling connections up behind it.
func TestActivatorWakeTimeout(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	old := wakeWaitBound
	wakeWaitBound = 100 * time.Millisecond
	t.Cleanup(func() { wakeWaitBound = old })

	gate := make(chan struct{})
	b.startGate = gate
	t.Cleanup(func() { close(gate) }) // let the stuck wake finish at test end

	_, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if !errors.Is(err, errWakeTimeout) {
		t.Fatalf("got %v, want errWakeTimeout", err)
	}
}

// Lease contention surfaces through the edge as 503 + Retry-After, and a
// failed wake leaves no entry behind (the next request retries the claim).
func TestActivatorLeaseContention503(t *testing.T) {
	a, b, _, reg := testActivator(t, time.Minute)
	b.acquireErr = &errLeaseHeld{holder: "old-incarnation", retryAfter: 7 * time.Second}

	if err := reg.Upsert(registry.Route{Host: "held.example.com", Kind: "instance", Socket: "/tmp/held.sock", Hibernate: true}); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)

	req, _ := http.NewRequest("GET", front.URL+"/", nil)
	req.Host = "held.example.com"
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status %d, want 503", resp.StatusCode)
	}
	if ra := resp.Header.Get("Retry-After"); ra != "8" { // 7s truncated + 1
		t.Fatalf("Retry-After %q, want 8", ra)
	}
	if a.entryCount() != 0 {
		t.Fatal("failed wake left an entry behind")
	}
}

// Per-host gating (the opt-in): with the activator CONFIGURED, only a
// hibernate-flagged instance goes through the wake path; an unflagged one is
// proxied straight, exactly as if there were no activator at all — the
// deploy-safety property that lets the bucket env ship fleet-wide before any
// host is flagged.
func TestRouterPerHostHibernateGating(t *testing.T) {
	requireUnixSockets(t)
	a, b, _, reg := testActivator(t, time.Minute)

	// Two live backends playing the instances. The flagged host's listener
	// stands in for the child a wake would have spawned (the fake backend's
	// Start() spawns nothing).
	bind := func(name string) string {
		sock := tmpSock(t)
		ln, err := net.Listen("unix", sock)
		if err != nil {
			t.Fatalf("listen unix: %v", err)
		}
		srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			io.WriteString(w, name)
		})}
		go srv.Serve(ln)
		t.Cleanup(func() { srv.Close() })
		return sock
	}
	if err := reg.Upsert(registry.Route{Host: "hib.example.com", Kind: "instance", Socket: bind("hib"), Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "plain.example.com", Kind: "instance", Socket: bind("plain")}); err != nil {
		t.Fatal(err)
	}

	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)

	// The unflagged host: proxied straight, the activator never touched.
	if body, code := proxyGet(t, front.URL, "plain.example.com"); code != 200 || body != "plain" {
		t.Fatalf("unflagged host: got %d %q, want 200 plain", code, body)
	}
	if acquires, _, starts := b.counts(); acquires != 0 || starts != 0 {
		t.Fatalf("unflagged instance touched the activator (acquires=%d starts=%d)", acquires, starts)
	}

	// The flagged host: served through a full wake — lease, restore, spawn.
	if body, code := proxyGet(t, front.URL, "hib.example.com"); code != 200 || body != "hib" {
		t.Fatalf("flagged host: got %d %q, want 200 hib", code, body)
	}
	if acquires, restores, starts := b.counts(); acquires != 1 || restores != 1 || starts != 1 {
		t.Fatalf("flagged host skipped the wake path: acquires=%d restores=%d starts=%d (want 1 each)", acquires, restores, starts)
	}
}

// The activator's own gate: handed an unflagged (or non-instance) route it
// declines — returns false, writes nothing, touches no backend — so the edge
// proxies straight. This is the bool half of the edge.Activator contract.
func TestServeHTTPUnmanagedRouteReturnsFalse(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	noop := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})

	rec := httptest.NewRecorder()
	req := httptest.NewRequest("GET", "http://plain.example.com/", nil)
	unflagged := registry.Route{Host: "plain.example.com", Kind: "instance", Socket: "/tmp/x.sock"}
	if a.ServeHTTP(rec, req, unflagged, noop) {
		t.Fatal("unflagged instance route was handled by the activator")
	}
	svc := registry.Route{Host: "svc.example.com", Kind: "service", Socket: "/tmp/s.sock", Hibernate: true}
	if a.ServeHTTP(rec, req, svc, noop) {
		t.Fatal("service route was handled by the activator")
	}
	if rec.Body.Len() != 0 {
		t.Fatalf("declined routes wrote a response: %q", rec.Body.String())
	}
	if acquires, _, starts := b.counts(); acquires != 0 || starts != 0 {
		t.Fatalf("declined routes touched the backend (acquires=%d starts=%d)", acquires, starts)
	}
}

// A scheduled tick must not move lastActive: it is the platform's own clock
// poking the instance, not a visitor. A tick that counted would make any
// schedule tighter than the idle window a permanent keep-awake.
func TestServeScheduledDoesNotStampActivity(t *testing.T) {
	const host = "tick.bab.oncarlos.com"
	a, _, clk, _ := testActivator(t, time.Hour)
	rt := testRoute(host)
	arrangeAwake(t, a, rt) // a real beginRequest round trip, then release

	a.mu.Lock()
	before := a.entries[rt.Host].lastActive
	a.mu.Unlock()

	clk.advance(time.Minute)

	rec := httptest.NewRecorder()
	req := httptest.NewRequest("POST", "http://carlos/jobs/sweep", nil)
	if !a.ServeScheduled(rec, req, rt, okHandler()) {
		t.Fatal("ServeScheduled returned false for a managed route")
	}

	a.mu.Lock()
	after := a.entries[rt.Host].lastActive
	inflight := a.entries[rt.Host].inflight
	a.mu.Unlock()

	if !after.Equal(before) {
		t.Errorf("lastActive moved %v -> %v; a scheduled tick must not count as activity", before, after)
	}
	if inflight != 0 {
		t.Errorf("inflight = %d after delivery, want 0 (the hold is released even though the stamp is not written)", inflight)
	}
}

// The other half of the same coin, asserted against the first so the two
// paths can never quietly become one again — which is the bug this fixes.
func TestServeHTTPStillStampsActivity(t *testing.T) {
	const host = "visitor.bab.oncarlos.com"
	a, _, clk, _ := testActivator(t, time.Hour)
	rt := testRoute(host)
	arrangeAwake(t, a, rt)

	a.mu.Lock()
	before := a.entries[rt.Host].lastActive
	a.mu.Unlock()

	clk.advance(time.Minute)

	rec := httptest.NewRecorder()
	req := httptest.NewRequest("GET", "http://carlos/", nil)
	if !a.ServeHTTP(rec, req, rt, okHandler()) {
		t.Fatal("ServeHTTP returned false for a managed route")
	}

	a.mu.Lock()
	after := a.entries[rt.Host].lastActive
	a.mu.Unlock()

	if !after.After(before) {
		t.Errorf("lastActive did not move on a visitor request (%v -> %v)", before, after)
	}
}

// A tick is a wake source and must stay one: an instance asleep when its
// timer comes due has to be woken by the delivery itself.
func TestServeScheduledWakesASleepingInstance(t *testing.T) {
	const host = "asleep.bab.oncarlos.com"
	a, _, _, _ := testActivator(t, time.Hour) // no entries yet: the host is asleep
	rt := testRoute(host)

	rec := httptest.NewRecorder()
	req := httptest.NewRequest("POST", "http://carlos/jobs/sweep", nil)
	if !a.ServeScheduled(rec, req, rt, okHandler()) {
		t.Fatal("ServeScheduled returned false for a managed route")
	}
	if rec.Code != http.StatusOK {
		t.Fatalf("status %d, want 200 — the tick must wake the instance, not be refused", rec.Code)
	}
	a.mu.Lock()
	e := a.entries[rt.Host]
	a.mu.Unlock()
	if e == nil || e.state != actAwake {
		t.Fatal("instance is not awake after a scheduled delivery")
	}
}

// A tick never wakes a route the box has stopped serving (G12): the move
// cut sets "moving", and a tick that woke the host there would restore it
// and let it write behind the move's final mirror round. This gate is the
// half that survives an agent restart, when the in-memory fence is gone.
func TestServeScheduledRefusesARouteThatIsNotReady(t *testing.T) {
	const host = "moving.bab.oncarlos.com"
	a, _, _, _ := testActivator(t, time.Hour)
	rt := testRoute(host)
	rt.Status = "moving"

	rec := httptest.NewRecorder()
	req := httptest.NewRequest("POST", "http://carlos/jobs/sweep", nil)
	if !a.ServeScheduled(rec, req, rt, okHandler()) {
		t.Fatal("ServeScheduled returned false: the runner would deliver to the socket itself")
	}
	if rec.Code != http.StatusServiceUnavailable {
		t.Fatalf("status %d, want 503", rec.Code)
	}
	a.mu.Lock()
	e := a.entries[rt.Host]
	a.mu.Unlock()
	if e != nil {
		t.Fatal("a tick woke a moving route")
	}
}

// LiftFence reopens only the fence it names: a stale generation leaves a
// newer fence closed, and the right one lets the next tick wake the host.
func TestLiftFenceReopensOnlyItsOwnGeneration(t *testing.T) {
	const host = "fenced.bab.oncarlos.com"
	a, _, _, _ := testActivator(t, time.Hour)
	rt := testRoute(host)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := a.FenceAndStop(ctx, host, 7); err != nil && !errors.Is(err, ErrStopUnproven) {
		t.Fatalf("FenceAndStop = %v", err)
	}
	if a.LiftFence(host, 6) || a.LiftFence(host, 0) {
		t.Fatal("a stale generation lifted the fence")
	}
	rec := httptest.NewRecorder()
	a.ServeScheduled(rec, httptest.NewRequest("POST", "http://carlos/jobs/sweep", nil), rt, okHandler())
	if rec.Code != http.StatusServiceUnavailable {
		t.Fatalf("fenced tick status %d, want 503", rec.Code)
	}
	if !a.LiftFence(host, 7) {
		t.Fatal("the fence's own generation did not lift it")
	}
	rec = httptest.NewRecorder()
	a.ServeScheduled(rec, httptest.NewRequest("POST", "http://carlos/jobs/sweep", nil), rt, okHandler())
	if rec.Code != http.StatusOK {
		t.Fatalf("tick after the lift: status %d, want 200", rec.Code)
	}
}

// The guard: a host whose next platform tick is nearer than the idle window
// stays resident. Sleeping would buy a few minutes of absence at the price of
// a full checkpoint-and-restore round trip, and then an immediate wake.
func TestSweepKeepsHostAwakeWhenATickIsImminent(t *testing.T) {
	const host = "imminent.bab.oncarlos.com"
	a, _, clk, reg := testActivator(t, 15*time.Minute)
	rt := testRoute(host)
	arrangeAwake(t, a, rt)

	// Arm 21 minutes out, then advance the clock 16 minutes: at sweep time
	// the tick is 5 minutes in the FUTURE — genuinely inside the window, not
	// merely overdue (that boundary is TestSweepKeepsHostAwakeForAnOverdueTick's
	// job). A guard that only checked next <= now would wrongly hibernate here.
	armTimer(t, reg, host, clk.now().Add(21*time.Minute))
	clk.advance(16 * time.Minute) // lastActive is now well past the window

	a.sweepOnce()

	a.mu.Lock()
	e := a.entries[rt.Host]
	a.mu.Unlock()
	if e == nil || e.state != actAwake {
		t.Fatal("host hibernated with a tick due inside the idle window")
	}
}

// Beyond the window, the sleep is worth taking.
func TestSweepHibernatesWhenTheNextTickIsFarOff(t *testing.T) {
	t.Parallel()
	const host = "faroff.bab.oncarlos.com"
	a, _, clk, reg := testActivator(t, 15*time.Minute)
	rt := testRoute(host)
	arrangeAwake(t, a, rt)

	// Next tick in an hour: well beyond the window.
	armTimer(t, reg, host, clk.now().Add(time.Hour))
	clk.advance(16 * time.Minute)

	a.sweepOnce()

	waitFor(t, "the host to hibernate with its next tick an hour away", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.entries[rt.Host] == nil
	})
}

// A host with no timer rows is untouched by the guard — the overwhelming
// majority of instances, and the case that must not regress.
func TestSweepHibernatesAHostWithNoTimers(t *testing.T) {
	t.Parallel()
	const host = "untimed.bab.oncarlos.com"
	a, _, clk, _ := testActivator(t, 15*time.Minute) // no timer rows at all
	rt := testRoute(host)
	arrangeAwake(t, a, rt)
	clk.advance(16 * time.Minute)

	a.sweepOnce()

	waitFor(t, "a host with no timers to hibernate as it always did", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		return a.entries[rt.Host] == nil
	})
}

// An overdue row (next_at already in the past — the runner has not got to it
// yet) is the most imminent tick there is, so it holds the host too.
func TestSweepKeepsHostAwakeForAnOverdueTick(t *testing.T) {
	const host = "overdue.bab.oncarlos.com"
	a, _, clk, reg := testActivator(t, 15*time.Minute)
	rt := testRoute(host)
	arrangeAwake(t, a, rt)

	// next_at already behind us: the runner will fire it within 15s.
	armTimer(t, reg, host, clk.now().Add(-time.Minute))
	clk.advance(16 * time.Minute)

	a.sweepOnce()

	a.mu.Lock()
	e := a.entries[rt.Host]
	a.mu.Unlock()
	if e == nil || e.state != actAwake {
		t.Fatal("host hibernated with an overdue tick pending")
	}
}

// A host held awake by the guard never reaches a "next wake" to converge on
// — the same gap the always-awake sidecar case closes. The guard's continue
// must fall through to the convergence check, not skip it, or a `carlos
// ship` behind a tick-held host is served stale indefinitely.
func TestSweepHeldForTickStillConverges(t *testing.T) {
	t.Parallel()
	const host = "tickheld.bab.oncarlos.com"
	a, b, clk, reg := testActivator(t, 15*time.Minute)
	var ver atomic.Value // the live/<host> symlink target; swapped to model a ship
	ver.Store("v1")
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	b.execVerFn = func() string { return ver.Load().(string) }

	rt := testRoute(host)
	arrangeAwake(t, a, rt)

	// A tight schedule: next tick always inside the window, so the guard
	// holds the host awake on every sweep from here on (an overdue row is
	// still "inside the window" — more so, if anything).
	armTimer(t, reg, host, clk.now().Add(5*time.Minute))
	clk.advance(16 * time.Minute) // idle timeout reached

	a.sweepOnce()
	a.mu.Lock()
	e := a.entries[rt.Host]
	a.mu.Unlock()
	if e == nil || e.state != actAwake {
		t.Fatal("host hibernated despite an imminent tick — guard not exercised")
	}

	// Ship: repoint the symlink while the guard still holds the host awake.
	ver.Store("v2")
	a.sweepOnce() // starts the grace clock only
	if a.entryCount() != 1 {
		t.Fatal("cycled inside the grace window")
	}

	// Past the grace window: the guard still holds (the tick is now even
	// more overdue), but convergence must still fire — that's the fix.
	clk.advance(ConvergeGrace + time.Second)
	a.sweepOnce()
	waitFor(t, "a tick-held host to still converge onto the new build", func() bool {
		return a.entryCount() == 0
	})
	if !b.lastProc().isStopped() {
		t.Fatal("stale instance's process was not stopped")
	}
}

// A reload that finds an awake host no longer hibernate-flagged retires its
// child — stopped, final-checkpointed (replication closed), entry forgotten —
// the edge-process half of flipping hibernation off. And once the flag is
// off, a recorded next_due no longer wakes it: the due index only acts on
// hibernating hosts, so a systemd-managed instance can never be spawned
// against its own unit.
func TestActivatorReloadRetiresUnflaggedHost(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, time.Hour)
	route := testRoute("flip.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatal(err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	// The retire's sleep re-records next_due AFTER the flag is cleared; an
	// already-past value must stay inert under the flag filter.
	b.setDue(clk.now().Add(-5 * time.Second).Unix())

	if err := reg.SetHibernate("flip.example.com", false); err != nil {
		t.Fatal(err)
	}
	if err := rt.reload(); err != nil {
		t.Fatalf("reload: %v", err)
	}
	waitFor(t, "retire", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("retired child was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("retired child's replication was not closed (no final checkpoint/lease release)")
	}

	// The due sweeper must not resurrect a host that's back under systemd.
	a.sweepOnce()
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("starts=%d after retire, want 1 (no due wake for a systemd-managed host)", starts)
	}
}

// A retire that lands mid-wake can't stop the child yet — it finishes the job
// once the wake resolves, so a flip-off during the small wake window still
// releases cleanly.
func TestActivatorRetireDuringWake(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, time.Hour)
	route := testRoute("mid.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatal(err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}

	// Start a request-less (due-time) wake and hold it in the waking state.
	b.startGate = make(chan struct{})
	if err := reg.SetNextDue("mid.example.com", clk.now().Add(-time.Second).Unix()); err != nil {
		t.Fatal(err)
	}
	a.sweepOnce()
	waitFor(t, "the wake to begin", func() bool { _, _, s := b.counts(); return s == 1 })

	// Flip off mid-wake: the reload's retire must defer, not deadlock.
	if err := reg.SetHibernate("mid.example.com", false); err != nil {
		t.Fatal(err)
	}
	if err := rt.reload(); err != nil {
		t.Fatalf("reload: %v", err)
	}
	close(b.startGate)

	waitFor(t, "post-wake retire", func() bool { return a.entryCount() == 0 })
	if !b.lastProc().isStopped() {
		t.Fatal("child woken mid-retire was not stopped")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("child woken mid-retire left replication open (lease never released)")
	}
}

// With hibernation on, sibling services (kind "service") are untouched: the
// edge proxies them directly, no wake, no activity tracking.
func TestActivatorServiceRoutesBypass(t *testing.T) {
	t.Parallel()
	requireUnixSockets(t)
	sock := tmpSock(t)
	ln, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatalf("listen unix: %v", err)
	}
	backend := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		io.WriteString(w, "service ok")
	})}
	go backend.Serve(ln)
	t.Cleanup(func() { backend.Close() })

	a, b, _, _ := testActivator(t, time.Minute)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	if err := reg.Upsert(registry.Route{Host: "svc.example.com", Kind: "service", Socket: sock}); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)

	req, _ := http.NewRequest("GET", front.URL+"/", nil)
	req.Host = "svc.example.com"
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	body, _ := io.ReadAll(resp.Body)
	resp.Body.Close()
	if resp.StatusCode != 200 || string(body) != "service ok" {
		t.Fatalf("got %d %q, want 200 service ok", resp.StatusCode, body)
	}
	if acquires, _, _ := b.counts(); acquires != 0 {
		t.Fatal("a service route touched the activator")
	}
}

// Shutdown checkpoints and stops every awake child before the edge exits.
func TestActivatorShutdownStopsChildren(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	for _, h := range []string{"one.example.com", "two.example.com"} {
		release, err := a.beginRequest(context.Background(), testRoute(h))
		if err != nil {
			t.Fatalf("beginRequest %s: %v", h, err)
		}
		release()
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)

	if a.entryCount() != 0 {
		t.Fatal("shutdown left entries behind")
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	for i, p := range b.procs {
		if !p.stopped {
			t.Fatalf("proc %d not stopped on shutdown", i)
		}
	}
	for i, r := range b.reps {
		if !r.closed {
			t.Fatalf("replication %d not closed on shutdown", i)
		}
	}
}

// Bounce stops a running instance so its next wake respawns with fresh
// env (config delivery). It must be a no-op for unknown hosts and for
// entries not currently awake — never a spurious wake, never a panic.
func TestBounceStopsAwakeInstanceOnly(t *testing.T) {
	a := newTestServiceAwake(t, "b.example.com")
	if !a.Bounce("b.example.com") {
		t.Fatal("Bounce(awake host) = false, want true")
	}
	a.mu.Lock()
	st := a.entries["b.example.com"].state
	a.mu.Unlock()
	if st != actStopping {
		t.Fatalf("state after Bounce = %v, want actStopping", st)
	}
	if a.Bounce("b.example.com") {
		t.Fatal("Bounce(already stopping) = true, want false")
	}
	if a.Bounce("never-woken.example.com") {
		t.Fatal("Bounce(unknown host) = true, want false")
	}
}

// Owns is the external-socket advisory's predicate (issue #17): a live entry
// in ANY state means whatever serves the host's socket is this activator's own
// child, so a host draining its SIGTERM must still read as owned — otherwise
// the advisory calls the activator's own children strays.
func TestOwnsCoversEntryLifetimeAnyState(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)
	host := "own.example.com"
	if a.Owns(host) {
		t.Fatal("Owns(host) = true before any request, want false")
	}

	release, err := a.beginRequest(context.Background(), testRoute(host))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	if !a.entryAwake(host) || !a.Owns(host) {
		t.Fatalf("awake: entryAwake=%v Owns=%v, want both true", a.entryAwake(host), a.Owns(host))
	}

	// Hold the stop open so the entry sits in stopping.
	gate := make(chan struct{})
	p := b.lastProc()
	p.mu.Lock()
	p.stopGate = gate
	p.mu.Unlock()
	clk.advance(time.Minute)
	a.sweepOnce()
	if a.entryAwake(host) {
		t.Fatal("entry still awake after the idle sweep, want stopping")
	}
	if !a.Owns(host) {
		t.Error("Owns(stopping host) = false, want true")
	}
	close(gate)

	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
	if a.Owns(host) {
		t.Error("Owns(host) = true after the entry was retired, want false")
	}
	if a.Owns("never-woken.example.com") {
		t.Error("Owns(unknown host) = true, want false")
	}
}

// --- due-time index --------------------------------------------------------------

// entryAwake reports whether host has a live entry in the awake state — how the
// due tests observe an async due-time wake resolving.
// entryRespawning: whether a converge respawn currently owns this entry's
// child — the interlock the sweeper honours.
func (a *Service) entryRespawning(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	return e != nil && e.workerRespawning
}

// entrySpawnVersion: the build the host's live child was exec'd from, "" if
// there is no entry — how a test sees that a respawn actually landed on the
// new binary rather than merely happening.
func (a *Service) entrySpawnVersion(host string) string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if e := a.entries[host]; e != nil {
		return e.spawnVersion
	}
	return ""
}

func (a *Service) entryAwake(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	e := a.entries[host]
	return e != nil && e.state == actAwake
}

// The full self-maintaining cycle: sleep records the instance's next due time
// in the registry; the sweeper wakes the host (no request) when it arrives; the
// following sleep re-records — and once nothing is scheduled, no more due wakes.
func TestActivatorDueTimeWakeCycle(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	route := testRoute("due.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatalf("upsert: %v", err)
	}

	// Wake by request; the instance has a delivery scheduled 30s out.
	dueAt := clk.now().Add(30 * time.Second).Unix()
	b.setDue(dueAt)
	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	// Idle out: the sleep probes the instance and records its due time.
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
	if rt, err := reg.Get(route.Host); err != nil || rt == nil || rt.NextDue != dueAt {
		t.Fatalf("registry next_due = %+v, %v; want %d", rt, err, dueAt)
	}

	// Before the due time: sweeps leave it asleep.
	a.sweepOnce()
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("woke before its due time (starts=%d)", starts)
	}

	// At the due time: the sweeper wakes it with no client request.
	clk.advance(20 * time.Second) // now = wake+31s ≥ due
	a.sweepOnce()
	waitFor(t, "due-time wake", func() bool { return a.entryAwake(route.Host) })
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d after due wake, want 2", starts)
	}

	// The woken instance stays up for the normal idle window, then sleeps and
	// re-records — nothing scheduled now, so the index clears and stays quiet.
	b.setDue(0)
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "re-hibernation", func() bool { return a.entryCount() == 0 })
	if rt, _ := reg.Get(route.Host); rt.NextDue != 0 {
		t.Fatalf("next_due = %d after empty-schedule sleep, want 0", rt.NextDue)
	}
	clk.advance(time.Hour)
	a.sweepOnce()
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d with nothing scheduled, want 2 (no phantom due wake)", starts)
	}
}

// A due time already in the past at sleep time (a delivery that raced the stop)
// is recorded as-is: the sweeper wakes the host on its next tick — once — and
// there is no ping-pong, because a woken instance can't sleep again until a
// full idle window passes.
func TestActivatorDuePastWakesOnceNoPingPong(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	route := testRoute("late.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	b.setDue(clk.now().Add(-5 * time.Second).Unix()) // overdue before it even sleeps

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })

	// Next tick: woken for the overdue delivery.
	a.sweepOnce()
	waitFor(t, "overdue wake", func() bool { return a.entryAwake(route.Host) })

	// No thrash: with the wake fresh (clock unmoved), repeated sweeps neither
	// re-sleep it nor double-wake it, even though next_due is still in the past.
	for i := 0; i < 5; i++ {
		a.sweepOnce()
	}
	if a.entryCount() != 1 || !a.entryAwake(route.Host) {
		t.Fatal("due-woken instance did not stay awake through its idle window")
	}
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts=%d, want 2 (request wake + one due wake)", starts)
	}
}

// A failed next-due probe (dead socket, missing admin token, …) records 0 and
// the sleep proceeds — the instance degrades to "sleeps until a request",
// never a crash and never a stuck stop.
func TestActivatorDueProbeFailureRecordsZero(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	route := testRoute("deaf.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	b.dueErr = errors.New("dial unix: no such file")

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation despite probe failure", func() bool { return a.entryCount() == 0 })

	if rt, _ := reg.Get(route.Host); rt.NextDue != 0 {
		t.Fatalf("next_due = %d after failed probe, want 0", rt.NextDue)
	}
	clk.advance(time.Hour)
	a.sweepOnce()
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("starts=%d, want 1 (a failed probe must not invent a wake)", starts)
	}
}

// The index lives in the registry, so a due time recorded before an edge
// restart still fires: a brand-new activator over the same registry — no
// in-memory state carried over — wakes the host when its time arrives.
func TestActivatorDueSurvivesRestart(t *testing.T) {
	t.Parallel()
	a1, b1, clk, reg := testActivator(t, 10*time.Second)
	route := testRoute("reborn.example.com")
	if err := reg.Upsert(route); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	b1.setDue(clk.now().Add(30 * time.Second).Unix())

	release, err := a1.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	clk.advance(11 * time.Second)
	a1.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a1.entryCount() == 0 })

	// "Restart": a fresh activator, same registry.
	b2 := &fakeBackend{}
	a2 := New(Config{Bucket: "test-bucket", Idle: 10 * time.Second}, b2, reg)
	a2.now = clk.now

	clk.advance(25 * time.Second) // past the due time
	a2.sweepOnce()
	waitFor(t, "post-restart due wake", func() bool { return a2.entryAwake(route.Host) })
	if _, _, starts := b2.counts(); starts != 1 {
		t.Fatalf("starts=%d on the new activator, want 1", starts)
	}
}

// --- sidecar co-spawn ----------------------------------------------------------

// indexOf: position of ev in the backend's order log (-1 = absent).
func indexOf(events []string, ev string) int {
	for i, e := range events {
		if e == ev {
			return i
		}
	}
	return -1
}

// A host with a sidecar staged gets it co-spawned on wake, strictly AFTER the
// instance process is up — the sidecar dials the instance, so spawning it
// first would just burn its reconnect backoff. (Every other test in this file
// runs with sidecarStaged=false and asserts nothing sidecar-ish happens, which
// is the "host without a sidecar behaves exactly as today" half.)
func TestActivatorCoSpawnsSidecarAfterInstance(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	b.sidecarStaged = true

	release, err := a.beginRequest(context.Background(), testRoute("bots.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	if got := b.sidecarStartCount(); got != 1 {
		t.Fatalf("sidecar starts = %d, want 1", got)
	}
	ev := b.events()
	if i, s := indexOf(ev, "start:instance"), indexOf(ev, "start:sidecar"); i == -1 || s == -1 || s < i {
		t.Fatalf("sidecar did not start strictly after the instance: %v", ev)
	}
}

// A sidecar that fails to spawn must not fail the wake — members > bots: the
// host serves without its sidecar, and the sweeper keeps retrying with backoff
// until a spawn sticks.
func TestActivatorSidecarFailureDoesNotFailWake(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	b.sidecarStaged = true
	b.setSidecarErr(errors.New("exec: live symlink gone"))
	route := testRoute("botless.example.com")

	release, err := a.beginRequest(context.Background(), route)
	if err != nil {
		t.Fatalf("beginRequest with a failing sidecar: %v", err)
	}
	release()
	if !a.entryAwake(route.Host) {
		t.Fatal("instance not awake after a sidecar-only failure")
	}
	if got := b.sidecarStartCount(); got != 1 {
		t.Fatalf("sidecar attempts = %d, want 1 (the wake's)", got)
	}

	// The sweeper retries: attempt 2 immediately (first retry), then gated by
	// the doubling backoff; once the fault clears, a retry lands the sidecar.
	// The restart call itself now runs off a.mu (design §6.7), so a fresh
	// attempt lands asynchronously. waitFor gives it a moment rather than
	// asserting on sweepOnce's own (now much shorter) return.
	a.sweepOnce()
	waitFor(t, "sidecar restart attempt 2 to land", func() bool { return b.sidecarStartCount() == 2 && a.sidecarSettled(route.Host) })
	if got := b.sidecarStartCount(); got != 2 {
		t.Fatalf("sidecar attempts after first sweep = %d, want 2", got)
	}
	a.sweepOnce() // still inside the 2s gate — no attempt
	if got := b.sidecarStartCount(); got != 2 {
		t.Fatalf("backoff gate ignored: attempts = %d, want 2", got)
	}
	b.setSidecarErr(nil)
	clk.advance(3 * time.Second)
	a.sweepOnce()
	waitFor(t, "sidecar restart attempt 3 to land", func() bool { return b.sidecarStartCount() == 3 && a.sidecarSettled(route.Host) })
	if got := b.sidecarStartCount(); got != 3 {
		t.Fatalf("sidecar attempts after fault cleared = %d, want 3", got)
	}
	if b.lastSidecar().isStopped() || b.lastSidecar().Exited() {
		t.Fatal("recovered sidecar isn't running")
	}
}

// Co-stop ordering at sleep: the sidecar is stopped FIRST — before the
// next-due probe (so bots can't shift the schedule being recorded) and before
// the instance's own stop (so the sidecar never outlives its instance).
func TestActivatorCoStopOrdering(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)
	b.sidecarStaged = true

	release, err := a.beginRequest(context.Background(), testRoute("bots.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })

	if !b.lastSidecar().isStopped() {
		t.Fatal("sidecar not stopped at sleep")
	}
	ev := b.events()
	sc, due, inst := indexOf(ev, "stop:sidecar"), indexOf(ev, "nextDue"), indexOf(ev, "stop:instance")
	if sc == -1 || due == -1 || inst == -1 || !(sc < due && due < inst) {
		t.Fatalf("co-stop order wrong: %v (want stop:sidecar < nextDue < stop:instance)", ev)
	}
}

// Orphan prevention: whatever takes the instance down — here a crashed child,
// but a lost lease and a retire ride the identical stop path — takes the
// sidecar with it. An orphaned sidecar would reconnect-loop against the edge
// forever (503s at best; a wake source if the marker rule ever broke).
func TestActivatorInstanceDeathStopsSidecar(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	b.sidecarStaged = true

	release, err := a.beginRequest(context.Background(), testRoute("bots.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	p := b.lastProc()
	p.mu.Lock()
	p.dead = true
	p.mu.Unlock()
	a.sweepOnce()
	waitFor(t, "stand-down", func() bool { return a.entryCount() == 0 })
	if !b.lastSidecar().isStopped() {
		t.Fatal("instance died but its sidecar was left running (orphan)")
	}
}

// Sidecar death while the instance stays awake: restarted with doubling,
// capped backoff — Restart=always semantics, sweep-driven — and a run that
// held for sidecarHealthyRun earns a fresh backoff.
func TestActivatorSidecarRestartBackoff(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	b.sidecarStaged = true

	release, err := a.beginRequest(context.Background(), testRoute("bots.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()
	kill := func() {
		sc := b.lastSidecar()
		sc.mu.Lock()
		sc.dead = true
		sc.mu.Unlock()
	}

	// The restart call itself runs off a.mu (design §6.7): each waitFor below
	// gives the previous attempt's goroutine a moment to land, both the
	// count and the entry's sidecarStarting flag, before the next kill/sweep,
	// so a slow scheduler can't make this sequence's next sweepOnce see the
	// prior attempt still "in flight" and skip its own.
	host := "bots.example.com"

	// T0: first crash — restarted on the next sweep (attempt 2), gate 2s.
	kill()
	a.sweepOnce()
	waitFor(t, "restart 2 to land", func() bool { return b.sidecarStartCount() == 2 && a.sidecarSettled(host) })
	if got := b.sidecarStartCount(); got != 2 {
		t.Fatalf("first restart: attempts = %d, want 2", got)
	}
	// Crash-loop: still inside the 2s gate — no restart.
	kill()
	a.sweepOnce()
	if got := b.sidecarStartCount(); got != 2 {
		t.Fatalf("gate ignored: attempts = %d, want 2", got)
	}
	// T3: past the 2s gate — attempt 3, gate doubles to 4s.
	clk.advance(3 * time.Second)
	a.sweepOnce()
	waitFor(t, "restart 3 to land", func() bool { return b.sidecarStartCount() == 3 && a.sidecarSettled(host) })
	if got := b.sidecarStartCount(); got != 3 {
		t.Fatalf("second restart: attempts = %d, want 3", got)
	}
	// T6: 3s later is inside the 4s gate — the doubling is real.
	kill()
	clk.advance(3 * time.Second)
	a.sweepOnce()
	if got := b.sidecarStartCount(); got != 3 {
		t.Fatalf("doubled gate ignored: attempts = %d, want 3", got)
	}
	// T8: past it — attempt 4 (gate now 8s).
	clk.advance(2 * time.Second)
	a.sweepOnce()
	waitFor(t, "restart 4 to land", func() bool { return b.sidecarStartCount() == 4 && a.sidecarSettled(host) })
	if got := b.sidecarStartCount(); got != 4 {
		t.Fatalf("third restart: attempts = %d, want 4", got)
	}
	// A healthy run (≥ sidecarHealthyRun) resets the ladder: the next crash
	// restarts immediately and its gate is back to the 2s base…
	clk.advance(2 * time.Minute)
	kill()
	a.sweepOnce()
	waitFor(t, "restart 5 to land", func() bool { return b.sidecarStartCount() == 5 && a.sidecarSettled(host) })
	if got := b.sidecarStartCount(); got != 5 {
		t.Fatalf("post-healthy restart: attempts = %d, want 5", got)
	}
	// …which a 3s advance clears (16s — the unreset ladder — would not).
	kill()
	clk.advance(3 * time.Second)
	a.sweepOnce()
	waitFor(t, "restart 6 to land", func() bool { return b.sidecarStartCount() == 6 && a.sidecarSettled(host) })
	if got := b.sidecarStartCount(); got != 6 {
		t.Fatalf("backoff not reset after a healthy run: attempts = %d, want 6", got)
	}
}

// The activity question: the co-spawned sidecar dials its instance's public
// URL through this very edge (the box pins its hosted names to loopback), so
// without the marker its keepalives would be member activity and the pair
// could never sleep. Full front-door check: marked requests are served while
// awake — with the marker stripped before the backend — but stamp nothing
// (the host hibernates on schedule beneath them), and once asleep they get
// 503 and wake nothing. An outsider spoofing the marker gets exactly that
// same treatment: it can't wake the host and can't suppress anyone else's
// stamps (stamps are per-request), so the marker is not a DoS lever in either
// direction.
func TestActivatorSidecarRequestsInvisibleToActivity(t *testing.T) {
	t.Parallel()
	requireUnixSockets(t)
	a, b, clk, reg := testActivator(t, 10*time.Second)

	// A live backend listener stands in for the child a wake would have
	// spawned (the fake backend's Start() spawns nothing), recording whether
	// the sidecar marker ever leaks through the proxy.
	sock := tmpSock(t)
	ln, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatalf("listen unix: %v", err)
	}
	var leaked atomic.Bool
	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get(SidecarHeader) != "" {
			leaked.Store(true)
		}
		io.WriteString(w, "ok")
	})}
	go srv.Serve(ln)
	t.Cleanup(func() { srv.Close() })

	const host = "bots.example.com"
	if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Socket: sock, Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	rt, err := newTestRouter(reg, a)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	front := httptest.NewServer(rt)
	t.Cleanup(front.Close)

	markedGet := func() int {
		req, _ := http.NewRequest("GET", front.URL+"/", nil)
		req.Host = host
		req.Header.Set(SidecarHeader, "1")
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatalf("marked request: %v", err)
		}
		defer resp.Body.Close()
		io.Copy(io.Discard, resp.Body)
		return resp.StatusCode
	}

	// A member request wakes the host.
	if body, code := proxyGet(t, front.URL, host); code != 200 || body != "ok" {
		t.Fatalf("member wake: got %d %q", code, body)
	}
	// Marked requests are served while awake…
	if code := markedGet(); code != 200 {
		t.Fatalf("marked request while awake: %d, want 200", code)
	}
	// …but keep stamping nothing: sidecar traffic riding through the whole
	// idle window must not delay the sleep.
	clk.advance(6 * time.Second)
	if code := markedGet(); code != 200 {
		t.Fatalf("marked request mid-window: %d, want 200", code)
	}
	clk.advance(5 * time.Second) // 11s since the member request
	a.sweepOnce()
	waitFor(t, "hibernation beneath sidecar traffic", func() bool { return a.entryCount() == 0 })

	// Asleep: marked requests are refused and wake NOTHING — this is also the
	// outsider-spoof case, and it's harmless.
	if code := markedGet(); code != 503 {
		t.Fatalf("marked request while asleep: %d, want 503", code)
	}
	if a.entryCount() != 0 {
		t.Fatal("a marked request created an entry (a wake) for a sleeping host")
	}
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("starts = %d after marked requests, want 1 (no sidecar-triggered wake)", starts)
	}
	// An unmarked request still wakes it, and the marker never reached the
	// backend on any request.
	if _, code := proxyGet(t, front.URL, host); code != 200 {
		t.Fatalf("member re-wake: %d, want 200", code)
	}
	if _, _, starts := b.counts(); starts != 2 {
		t.Fatalf("starts = %d after member re-wake, want 2", starts)
	}
	if leaked.Load() {
		t.Fatal("the sidecar marker leaked through the proxy to the backend")
	}
}

// --- standalone sidecars (workers) -------------------------------------------
//
// A worker is a member of the same state machine with the request path removed:
// nothing can dial it, so every wake arrives from the due-time sweep, an
// operator command, or the agent's boot pass, and its only sign of life is
// contact over the agent socket. These tests pin exactly the places where its
// lifecycle DIVERGES from an instance's — the spawn call, the absent next-due
// probe, the two ceilings — and the places where it must not (lease release,
// DB parking, singleflight).

func TestWorkerWakeUsesStartWorkerAndNoProbe(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := b.events(); !contains(got, "startWorker:bot.x.internal") {
		t.Fatalf("events %v — a worker must spawn through StartWorker, never Start", got)
	}
	if b.startCount() != 0 {
		t.Fatal("Start (the socket path, with its health probe) must never be called for a worker")
	}
}

// The wake hands the child its agent-socket coordinates and the per-wake
// token — the child's only way to reach back, and the activator's only way to
// authorize it. WorkerToken must agree with what the child was told, or every
// call the sidecar makes is a 401.
func TestWorkerWakeThreadsTokenAndSocketEnv(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	tok := a.WorkerToken("bot.x.internal")
	if tok == "" {
		t.Fatal("WorkerToken is empty for an awake worker")
	}
	env := b.lastWorkerEnv()
	if env["CARLOS_SIDECAR_TOKEN"] != tok {
		t.Fatalf("child got token %q but WorkerToken reports %q — the socket would refuse its own child",
			env["CARLOS_SIDECAR_TOKEN"], tok)
	}
	if env["CARLOS_SIDECAR_HOST"] != "bot.x.internal" || env["CARLOS_SIDECAR_NAME"] != "bot" {
		t.Fatalf("child env = %v, want the host and its first label", env)
	}
	if env["CARLOS_AGENT_SOCKET"] == "" {
		t.Fatalf("child env = %v, want the agent socket path (the child dials, never the platform)", env)
	}
	if a.WorkerToken("nobody.x.internal") != "" {
		t.Fatal("WorkerToken must be empty for a host with no live worker")
	}
}

func TestWorkerSleepsWhenIdleAndRecordsNextDue(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if err := a.SetWorkerNextDue("bot.x.internal", 4242); err != nil {
		t.Fatalf("SetWorkerNextDue: %v", err)
	}

	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "worker hibernation", func() bool { return a.entryCount() == 0 })

	got, err := reg.Get("bot.x.internal")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.NextDue != 4242 {
		t.Fatalf("next_due = %d, want 4242 — a worker's next-due arrives over the agent socket while it is ALIVE, not from a probe of a process we are about to kill", got.NextDue)
	}
	if b.dueProbes() != 0 {
		t.Fatal("a worker must never be probed for next-due: it has no socket to answer on")
	}
	if !b.lastRep().isClosed() {
		t.Fatal("the lease must be released and the DB parked on a worker sleep, exactly as for an instance")
	}
	if !b.workers[len(b.workers)-1].isStopped() {
		t.Fatal("the worker child was not stopped")
	}
}

func TestWorkerContactDefersSleep(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, 10*time.Second)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	clk.advance(9 * time.Second)
	a.NoteWorkerContact("bot.x.internal")
	clk.advance(9 * time.Second)
	a.sweepOnce()
	// entryAwake, not entryCount: a condemned entry still counts until its
	// stop goroutine finishes, so a count here passes either way.
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("agent-socket contact must defer the idle clock — it is a worker's only sign of life")
	}
}

// The row's own idle window wins over the platform default: the ceilings ride
// on the route (Task 1) precisely so the sweeper never needs a bucket read to
// decide a worker has sat idle long enough.
func TestWorkerRouteIdleWindowOverridesPlatformDefault(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, time.Hour) // the platform default would never fire
	rt := testWorkerRoute("bot.x.internal")
	rt.IdleSeconds = 30
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	clk.advance(20 * time.Second)
	a.sweepOnce()
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("worker slept before its own idle window elapsed")
	}
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "row-scoped idle sleep", func() bool { return a.entryCount() == 0 })
}

func TestWorkerMaxRunStopsALoopingWorker(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, time.Hour) // idle window never fires
	rt := testWorkerRoute("bot.x.internal")
	rt.MaxRunSeconds = 60
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	for i := 0; i < 10; i++ { // a worker that keeps talking but never idles
		clk.advance(10 * time.Second)
		a.NoteWorkerContact("bot.x.internal")
	}
	a.sweepOnce()
	waitFor(t, "max_run stop", func() bool { return a.entryCount() == 0 })
}

// No ceiling declared (0) means no ceiling — an always-on worker must not be
// stopped by a default nobody asked for.
func TestWorkerWithoutMaxRunKeepsRunning(t *testing.T) {
	t.Parallel()
	a, _, clk, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	for i := 0; i < 10; i++ {
		clk.advance(time.Hour)
		a.NoteWorkerContact("bot.x.internal")
		a.sweepOnce()
	}
	if a.entryCount() != 1 {
		t.Fatal("a worker with max_run 0 was stopped by a ceiling it never declared")
	}
}

func TestSleepWorkerIsImmediate(t *testing.T) {
	t.Parallel()
	a, _, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	a.SleepWorker("bot.x.internal")
	waitFor(t, "idle-requested stop", func() bool { return a.entryCount() == 0 })
}

func TestWakeWorkerIsIdempotent(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	rt := testWorkerRoute("bot.x.internal")
	for i := 0; i < 3; i++ {
		if err := a.WakeWorker(rt); err != nil {
			t.Fatalf("WakeWorker #%d: %v", i, err)
		}
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d, want 1 — a wake for an already-awake worker is a no-op, not a second child", n)
	}
}

// An already-awake worker whose lease is gone is BROKEN, not up. WakeWorker
// must stand it down and wake it afresh — reporting success to a boot pass or
// an operator for a worker that cannot write is the one answer that makes a
// dead sidecar look healthy.
func TestWakeWorkerStandsDownALostLeaseAndRewakes(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	rt := testWorkerRoute("bot.x.internal")
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastRep().setErr(errors.New("lease renewal failed"))

	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("second WakeWorker: %v", err)
	}
	if n := b.workerStarts(); n != 2 {
		t.Fatalf("workerStarts = %d — a wake for a worker with a lost lease must recycle it, not report success", n)
	}
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("the re-wake did not leave the worker awake")
	}
	if acquires, _, _ := b.counts(); acquires != 2 {
		t.Fatalf("lease acquires = %d, want 2 — the stand-down must re-claim the lease", acquires)
	}
}

// Concurrent wakes coalesce on one entry, the same singleflight the request
// path gets: three goroutines, one child, one lease.
func TestWakeWorkerCoalescesConcurrentCallers(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	rt := testWorkerRoute("bot.x.internal")

	var wg sync.WaitGroup
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := a.WakeWorker(rt); err != nil {
				t.Errorf("WakeWorker: %v", err)
			}
		}()
	}
	wg.Wait()
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d, want 1 — concurrent wakes must coalesce", n)
	}
	if acquires, _, _ := b.counts(); acquires != 1 {
		t.Fatalf("lease acquires = %d, want 1", acquires)
	}
}

// Shutdown announces itself BEFORE it snapshots the entry map, so a wake
// dispatched after that point would insert an entry nothing is left to stop:
// an orphan child holding the tenant's lease past the agent's exit. A wake
// arriving mid-drain is refused outright.
func TestWakeWorkerRefusesOnceShutdownHasBegun(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	a.Shutdown(ctx)

	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err == nil {
		t.Fatal("WakeWorker succeeded during shutdown — its child would outlive the agent, holding the lease")
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("workerStarts = %d during shutdown, want 0", n)
	}
	if a.entryCount() != 0 {
		t.Fatal("a wake during shutdown left an entry behind")
	}
}

// A failed worker wake releases the lease and leaves nothing behind, exactly
// as a failed instance wake does — the error paths that must release a lease
// are the reason a worker reuses doWake rather than reimplementing it.
func TestWorkerWakeFailureReleasesLeaseAndClearsEntry(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	b.workerErr = errors.New("exec: no argv")

	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err == nil {
		t.Fatal("WakeWorker returned nil for a spawn that failed")
	}
	if a.entryCount() != 0 {
		t.Fatal("a failed worker wake left an entry behind")
	}
	if n := b.releaseCount(); n != 1 {
		t.Fatalf("lease releases = %d, want exactly 1 on a failed worker wake", n)
	}
}

// A worker is nobody's parent: the legacy co-spawn step must not run for one,
// even on a box where the staging file happens to exist.
func TestWorkerNeverCoSpawnsASidecar(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	b.sidecarStaged = true

	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	a.sweepOnce() // the sweep's mid-flight adoption must not pick one up either
	if n := b.sidecarStartCount(); n != 0 {
		t.Fatalf("sidecar starts = %d for a worker, want 0 — a worker is nobody's parent", n)
	}
}

// RetireExcept reconciles the EDGE's managed set — and edge.Reload refuses a
// sidecar row on kind before it builds that set, so a worker's host can never
// be in `keep`. Without an explicit exemption every edge reload (any route
// change at all) would stop every running standalone sidecar.
func TestRetireExceptLeavesWorkersAlone(t *testing.T) {
	t.Parallel()
	a, _, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	a.RetireExcept(map[string]bool{}) // an edge reload that manages nothing

	waitFor(t, "the instance to retire", func() bool { return !a.Owns("one.example.com") })
	if a.WorkerToken("bot.x.internal") == "" {
		t.Fatal("an edge reload retired a running worker — its host can never appear in the edge's managed set")
	}
}

// The due-time sweep is a sidecar's ONLY timer wake source (nothing can dial
// one), so a due sidecar row must come up through the worker path — StartWorker
// and no health probe — not the instance one.
func TestDueSweepWakesASidecarAsAWorker(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, time.Hour)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	if err := reg.SetNextDue(rt.Host, clk.now().Unix()-1); err != nil {
		t.Fatalf("set next due: %v", err)
	}

	a.sweepOnce()
	waitFor(t, "due worker wake", func() bool { return b.workerStarts() == 1 })
	if b.startCount() != 0 {
		t.Fatal("a due sidecar was woken through the instance path (Start + health probe)")
	}
}

// A FIRED timer is one-shot. The row's next_due is not rewritten by a
// worker's stop (it has no socket to probe), so unless the wake itself
// consumes the value that fired, DueInstances hands the same row back on
// EVERY sweep: wake, restore, idle, park, re-wake, forever — one lease
// acquire and one S3 restore/park per cycle, and nothing in the logs that
// reads as a failure.
func TestDueFiredWorkerWakeHappensExactlyOnce(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	if err := reg.SetNextDue(rt.Host, clk.now().Unix()-1); err != nil {
		t.Fatalf("set next due: %v", err)
	}

	a.sweepOnce()
	waitFor(t, "the due-fired wake", func() bool { return a.entryAwake(rt.Host) })
	got, err := reg.Get(rt.Host)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.NextDue != 0 {
		t.Fatalf("next_due = %d after the wake it fired, want 0 — a fired timer must be consumed by the wake it caused", got.NextDue)
	}

	// It idles out, parks, and STAYS parked: several more sweeps, no second
	// wake, no churn.
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
	for i := 0; i < 5; i++ {
		clk.advance(11 * time.Second)
		a.sweepOnce()
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d, want 1 — one due time woke the sidecar more than once (wake/park churn, every sweep, forever)", n)
	}
	if n := b.releaseCount(); n > 1 {
		t.Fatalf("leases released %d times for one due time — the sidecar is churning", n)
	}
}

// The other half of the same distinction: a sidecar that goes to sleep BEFORE
// its timer fires must keep its schedule. Consuming at STOP (rather than on
// the wake the timer caused) would delete the feature — a sleeping sidecar's
// next_due is its only way back up, since nothing can dial one.
func TestWorkerSleepingBeforeItsDueTimeStillWakesForIt(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	if err := a.WakeWorker(rt); err != nil { // an operator wake, not a due-fired one
		t.Fatalf("WakeWorker: %v", err)
	}
	due := clk.now().Add(time.Minute).Unix()
	if err := a.SetWorkerNextDue(rt.Host, due); err != nil { // "wake me in a minute"
		t.Fatalf("SetWorkerNextDue: %v", err)
	}

	clk.advance(11 * time.Second) // idles out well before the due time
	a.sweepOnce()
	waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
	got, err := reg.Get(rt.Host)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.NextDue != due {
		t.Fatalf("next_due = %d after a sleep BEFORE the due time, want %d — a schedule that has not fired must survive the sleep", got.NextDue, due)
	}

	clk.advance(time.Minute)
	a.sweepOnce()
	waitFor(t, "the scheduled wake", func() bool { return b.workerStarts() == 2 })
}

// A FAILED due-fired wake must consume NOTHING. The index is the retry — the
// due sweep's own contract for a wake that could not complete — and consuming
// on failure would convert harmless churn into the far worse bug: a sidecar
// whose only way back up has been erased, silently, by a wake that never
// happened.
func TestFailedDueFiredWakeConsumesNothing(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	fired := clk.now().Unix() - 1
	if err := reg.SetNextDue(rt.Host, fired); err != nil {
		t.Fatalf("set next due: %v", err)
	}
	b.mu.Lock()
	b.workerErr = errors.New("no build to exec") // the wake fails at the spawn
	b.mu.Unlock()

	a.sweepOnce()
	waitFor(t, "the failed wake to clear its entry", func() bool { return a.entryCount() == 0 })
	got, err := reg.Get(rt.Host)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.NextDue != fired {
		t.Fatalf("next_due = %d after a FAILED due wake, want %d — the schedule is the retry, and a sidecar whose next_due is erased by a wake that never happened has no way back up at all", got.NextDue, fired)
	}

	// The next sweep retries it, exactly as the due sweep's own comment says.
	a.sweepOnce()
	waitFor(t, "the retry", func() bool { return b.workerStarts() == 2 })

	// …and once a wake actually succeeds, THAT one consumes the schedule.
	b.mu.Lock()
	b.workerErr = nil
	b.mu.Unlock()
	a.sweepOnce()
	waitFor(t, "the successful wake", func() bool { return a.entryAwake(rt.Host) })
	if got, err := reg.Get(rt.Host); err != nil || got.NextDue != 0 {
		t.Fatalf("next_due = %+v (%v) after the wake that SUCCEEDED, want 0", got, err)
	}
}

// The consumption is conditional on the row still holding the value that
// fired: the child spawned by this very wake may post its NEXT due time over
// the agent socket during the settle window, and an unconditional clear would
// eat it — a bot that ran once and never again.
func TestDueFiredWakeKeepsANewNextDuePostedByTheChild(t *testing.T) {
	t.Parallel()
	a, b, clk, reg := testActivator(t, 10*time.Second)
	rt := testWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	if err := reg.SetNextDue(rt.Host, clk.now().Unix()-1); err != nil {
		t.Fatalf("set next due: %v", err)
	}
	next := clk.now().Add(time.Minute).Unix()
	b.mu.Lock()
	b.workerStartHook = func(r registry.Route) {
		// The child's own /v1/next-due, landing while the wake is still in
		// flight — before anything could consume the time that fired.
		if err := a.SetWorkerNextDue(r.Host, next); err != nil {
			t.Errorf("SetWorkerNextDue: %v", err)
		}
	}
	b.mu.Unlock()

	a.sweepOnce()
	waitFor(t, "the due-fired wake", func() bool { return a.entryAwake(rt.Host) })
	got, err := reg.Get(rt.Host)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if got.NextDue != next {
		t.Fatalf("next_due = %d, want %d — consuming the FIRED time must not eat the one the child posted during its wake", got.NextDue, next)
	}

	// And that schedule is honoured: park, then wake again when it arrives.
	b.mu.Lock()
	b.workerStartHook = nil
	b.mu.Unlock()
	clk.advance(11 * time.Second)
	a.sweepOnce()
	waitFor(t, "the worker to park", func() bool { return a.entryCount() == 0 })
	clk.advance(time.Minute)
	a.sweepOnce()
	waitFor(t, "the child's own scheduled wake", func() bool { return b.workerStarts() == 2 })
}

// --- the settle window --------------------------------------------------------

// noteReadyAsChild reports readiness the way the real child must: presenting
// the token it was handed in its env, which the agent socket (Task 7) matches
// against WorkerToken before accepting anything. Calling NoteWorkerReady
// directly would walk around the authorization the feature actually depends
// on — and hide a token that is not visible until the wake resolves, which
// would make this whole early-ready path dead code in production.
func noteReadyAsChild(t *testing.T, a *Service, b *fakeBackend, host, version string) {
	t.Helper()
	presented := b.lastWorkerEnv()["CARLOS_SIDECAR_TOKEN"]
	live := a.WorkerToken(host)
	if presented == "" || live == "" || presented != live {
		t.Fatalf("the agent socket would REFUSE this child's ready call: it presents %q, WorkerToken says %q", presented, live)
	}
	a.NoteWorkerReady(host, version)
}

// The child's very first call arrives while its wake is still resolving — that
// call is what ends the settle window — so its credential must be live then,
// not only once the entry is awake.
func TestWorkerTokenIsLiveDuringTheSettleWindow(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	old := sidecarSettle
	sidecarSettle = 5 * time.Second
	defer func() { sidecarSettle = old }()

	done := make(chan error, 1)
	go func() { done <- a.WakeWorker(testWorkerRoute("bot.x.internal")) }()
	waitFor(t, "the waking entry to appear", func() bool { return a.Owns("bot.x.internal") })
	waitFor(t, "the child to be spawned", func() bool { return b.workerStarts() == 1 })

	if tok := a.WorkerToken("bot.x.internal"); tok == "" {
		t.Fatal("no token during the settle window — the socket would refuse the /v1/ready call it is waiting for")
	}
	noteReadyAsChild(t, a, b, "bot.x.internal", "v9") // releases the wake
	if err := <-done; err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
}

// A worker binds nothing, so there is no socket to probe: readiness is the
// settle window. A child that exits INSIDE it never became a working sidecar —
// that is a failed wake, and it must give the lease back exactly as a failed
// instance start does, not sit in the map holding one.
func TestWorkerExitInSettleWindowFailsWakeAndReleasesLease(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	b.workerDeadOnStart = true

	err := a.WakeWorker(testWorkerRoute("bot.x.internal"))
	if err == nil {
		t.Fatal("a worker that died inside its settle window reported a successful wake")
	}
	if !errors.Is(err, errWorkerExitedEarly) {
		t.Fatalf("wake error = %v, want it to name the settle-window exit", err)
	}
	if a.entryCount() != 0 {
		t.Fatal("a failed settle left an entry behind")
	}
	if n := b.releaseCount(); n != 1 {
		t.Fatalf("lease releases = %d, want exactly 1", n)
	}
	if !b.lastWorker().isStopped() {
		t.Fatal("the half-started child was not stopped")
	}
	// The credential is published at mint time so the child can call
	// /v1/ready during the window — a wake that FAILED must still take it
	// away, which the entry's deletion does.
	if tok := a.WorkerToken("bot.x.internal"); tok != "" {
		t.Fatalf("WorkerToken = %q after a failed wake — a dead wake's credential must not stay live", tok)
	}
}

// A child that survives the window is up — no probe, no readiness call needed.
func TestWorkerSurvivingSettleWindowIsAwake(t *testing.T) {
	a, _, _, _ := testActivator(t, time.Hour)
	old := sidecarSettle
	sidecarSettle = 60 * time.Millisecond
	defer func() { sidecarSettle = old }()

	start := time.Now()
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if elapsed := time.Since(start); elapsed < 50*time.Millisecond {
		t.Fatalf("wake resolved in %s — a worker nobody vouched for must survive the whole settle window first", elapsed)
	}
	if a.WorkerToken("bot.x.internal") == "" {
		t.Fatal("the worker is not awake after surviving its settle window")
	}
}

// …and a child that reports ready ends the window early: a fast sidecar must
// not pay ten seconds of dead time on every wake.
func TestWorkerReadyEndsSettleWindowEarly(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	old := sidecarSettle
	sidecarSettle = 5 * time.Second
	defer func() { sidecarSettle = old }()

	done := make(chan error, 1)
	start := time.Now()
	go func() { done <- a.WakeWorker(testWorkerRoute("bot.x.internal")) }()

	waitFor(t, "the waking entry to appear", func() bool { return a.Owns("bot.x.internal") })
	// The entry appears before the wake has minted the token and spawned the
	// child; a child cannot report ready before it exists. Waiting only for
	// the entry raced the spawn under load (main 9c535162's CI).
	waitFor(t, "the child to be spawned", func() bool { return b.workerStarts() == 1 })
	noteReadyAsChild(t, a, b, "bot.x.internal", "v9")

	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("WakeWorker: %v", err)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("a ready report did not end the settle window — the wake waited it out")
	}
	if elapsed := time.Since(start); elapsed > 2*time.Second {
		t.Fatalf("wake took %s despite a ready report", elapsed)
	}
}

// --- always-on workers: supervision ------------------------------------------

func alwaysOnWorkerRoute(host string) registry.Route {
	rt := testWorkerRoute(host)
	rt.Hibernate = false // the record exists ⇒ the process runs
	return rt
}

// An always-on worker's exit is a CRASH, not a finished job: it is restarted
// in place, keeping its lease and replication — and the backoff GROWS, so a
// crash-looping bot paces itself instead of hot-looping the box.
func TestAlwaysOnWorkerCrashRestartsWithGrowingBackoff(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}

	// Crash 1: restarted at once, and the 2s delay is armed for the next.
	// The spawn runs off a.mu (a bucket materialize must never be held across
	// the global mutex), so the count is awaited, not read straight after.
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 2)
	if a.entryCount() != 1 {
		t.Fatal("the crashed worker's entry was torn down — an always-on worker restarts in place")
	}
	if n := b.releaseCount(); n != 0 {
		t.Fatalf("lease releases = %d — an in-place restart must not churn the lease", n)
	}

	// Crash 2 inside the backoff: NOT restarted yet. entryRespawning is set
	// synchronously under the sweep's own lock, so its absence right after the
	// sweep is a fact, not a race.
	b.lastWorker().kill()
	a.sweepOnce()
	if a.entryRespawning("bot.x.internal") || b.workerStarts() != 2 {
		t.Fatalf("workerStarts = %d — a crash inside the backoff must wait, not hot-loop", b.workerStarts())
	}
	clk.advance(2 * time.Second)
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 3)

	// Crash 3: the delay has DOUBLED — 2s is no longer enough.
	b.lastWorker().kill()
	clk.advance(2 * time.Second)
	a.sweepOnce()
	if a.entryRespawning("bot.x.internal") || b.workerStarts() != 3 {
		t.Fatalf("workerStarts = %d — the backoff must double after each crash, not stay at %s", b.workerStarts(), sidecarRestartBase)
	}
	clk.advance(3 * time.Second)
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 4)
}

// waitForRestart waits until a supervised respawn has both spawned its child
// and INSTALLED it — the install and the interlock clear happen in one
// critical section, so waiting on the pair is what makes the next kill land on
// the new child rather than the outgoing one.
func waitForRestart(t *testing.T, a *Service, b *fakeBackend, host string, want int) {
	t.Helper()
	waitFor(t, fmt.Sprintf("restart #%d of %s", want, host), func() bool {
		return b.workerStarts() == want && !a.entryRespawning(host)
	})
}

// A run that lasted sidecarHealthyRun earns a fresh backoff — one crash a day
// must never escalate to the cap (the co-spawn supervisor's rule).
func TestAlwaysOnWorkerHealthyRunResetsBackoff(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastWorker().kill()
	a.sweepOnce() // restart 1 — backoff now 2s
	waitForRestart(t, a, b, "bot.x.internal", 2)

	// The replacement runs healthily for well past the healthy-run mark, then
	// dies: the next restart is immediate, not delayed.
	clk.advance(sidecarHealthyRun + time.Minute)
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 3)
}

// A restart is a new incarnation: a fresh token, so the dead child's
// credential can never speak for the live one.
func TestAlwaysOnWorkerRestartMintsANewToken(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	first := a.WorkerToken("bot.x.internal")
	b.lastWorker().kill()
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 2)

	second := a.WorkerToken("bot.x.internal")
	if second == "" || second == first {
		t.Fatalf("token after restart = %q (was %q) — every incarnation gets its own", second, first)
	}
	if env := b.lastWorkerEnv(); env["CARLOS_SIDECAR_TOKEN"] != second {
		t.Fatalf("restarted child was handed %q but the socket expects %q", env["CARLOS_SIDECAR_TOKEN"], second)
	}
}

// A HIBERNATING worker's exit means "I finished my work" — the existing
// dead-child stand-down is exactly right, and must not become a restart loop.
func TestHibernatingWorkerExitSleepsRatherThanRestarting(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil { // Hibernate: true
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastWorker().kill() // a CLEAN exit: "my work is done"
	a.sweepOnce()

	waitFor(t, "the finished worker to sleep", func() bool { return a.entryCount() == 0 })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d — a hibernating worker that exited has FINISHED; it must not be restarted", n)
	}
	if !b.lastRep().isClosed() {
		t.Fatal("the lease was not released when the finished worker stopped")
	}
}

// An always-on worker has no idle clock: idleness is no reason to stop
// something whose record says it must be running, and a stopped one has no
// wake source at all (no request can reach it, and it has no schedule).
func TestAlwaysOnWorkerIgnoresTheIdleClock(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, 10*time.Second)
	rt := alwaysOnWorkerRoute("bot.x.internal")
	rt.IdleSeconds = 5
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}

	clk.advance(time.Hour) // far past the idle window
	a.sweepOnce()
	// entryAwake, not entryCount: beginStopLocked flips the state under the
	// lock and stops on its own goroutine, so a count taken right after the
	// sweep would still see the entry it just condemned.
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("an always-on worker was stopped by an idle window — nothing would ever wake it again")
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d, want 1 — nothing should have recycled a worker with no declared ceiling", n)
	}
}

// A HIBERNATING worker that exits NON-ZERO has crashed, and design §7.2 says
// a crash restarts, unqualified. Read as a finish, it would sleep forever:
// it fell over before it could set a next-due, and nothing can dial one.
func TestHibernatingWorkerCrashRestartsInsteadOfSleeping(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil { // Hibernate: true
		t.Fatalf("WakeWorker: %v", err)
	}
	b.lastWorker().crash() // fell over, non-zero
	a.sweepOnce()

	waitForRestart(t, a, b, "bot.x.internal", 2)
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("a crashed hibernating worker was parked — it never set a next-due, so nothing would wake it again")
	}
	if b.lastRep().isClosed() {
		t.Fatal("the lease was released on a crash restart — only the child died")
	}
}

// max_run is the wedged-reconnect-loop guard, and an always-on worker is
// exactly the shape that gets wedged. It is spent as a RECYCLE there, not a
// stop: stopping one leaves nothing to start it again.
func TestAlwaysOnWorkerMaxRunRecyclesItsChild(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Hour)
	rt := alwaysOnWorkerRoute("bot.x.internal")
	rt.MaxRunSeconds = 30
	if err := a.WakeWorker(rt); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	first := b.lastWorker()

	clk.advance(31 * time.Second)
	a.sweepOnce()
	waitForRestart(t, a, b, "bot.x.internal", 2)
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("an always-on worker was STOPPED by its ceiling — nothing would ever start it again")
	}
	if !first.isStopped() {
		t.Fatal("the wedged child was left running beside its replacement")
	}
	if b.lastRep().isClosed() {
		t.Fatal("the lease was released by a max_run recycle — the entry keeps its lease and replication")
	}

	// And the ceiling re-arms on the new child rather than firing every sweep.
	a.sweepOnce()
	if n := b.workerStarts(); n != 2 {
		t.Fatalf("workerStarts = %d, want 2 — the fresh child's own max_run clock starts at its spawn", n)
	}
}

// §7.2's contract is "the record exists ⇒ the process runs", and until the
// sweep grew this arm it held only across agent RESTARTS: the boot pass runs
// once, at arming, and the due-time loop gates on Hibernate — so an always-on
// sidecar created while the agent was running had nothing at all that would
// start it.
func TestSweepStartsAnAlwaysOnWorkerThatAppearedMidFlight(t *testing.T) {
	t.Parallel()
	a, b, _, reg := testActivator(t, time.Hour)
	rt := alwaysOnWorkerRoute("bot.x.internal")
	if err := reg.Upsert(rt); err != nil { // the reconcile pass mints it, mid-flight
		t.Fatalf("upsert: %v", err)
	}

	a.sweepOnce()
	waitFor(t, "the always-on sidecar to start", func() bool { return a.entryAwake(rt.Host) })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d, want 1", n)
	}

	// And it is a one-shot, not a per-tick respawn: an entry that is already
	// live is left entirely alone.
	for i := 0; i < 3; i++ {
		a.sweepOnce()
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d after three more sweeps, want 1 — a running always-on sidecar must not be started again", n)
	}
}

// The arm is for always-on rows only. A HIBERNATING sidecar exists precisely
// so that nothing runs until a wake — starting one here would delete the
// class.
func TestSweepNeverStartsAHibernatingWorker(t *testing.T) {
	a, b, _, reg := testActivator(t, time.Hour)
	rt := testWorkerRoute("bot.x.internal") // Hibernate: true, no next_due
	if err := reg.Upsert(rt); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	for i := 0; i < 3; i++ {
		a.sweepOnce()
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("workerStarts = %d — a hibernating sidecar must stay asleep until a wake source fires", n)
	}
	if a.entryCount() != 0 {
		t.Fatal("a hibernating sidecar was given a live entry by the sweep")
	}
}

// A stranded always-on worker is a bounded batch, not a stampede: each wake
// claims an S3 lease and restores a database.
func TestSweepBoundsAlwaysOnStartsPerTick(t *testing.T) {
	t.Parallel()
	a, b, _, reg := testActivator(t, time.Hour)
	for i := 0; i < maxWorkerStartsPerSweep+3; i++ {
		rt := alwaysOnWorkerRoute(fmt.Sprintf("bot%02d.x.internal", i))
		if err := reg.Upsert(rt); err != nil {
			t.Fatalf("upsert: %v", err)
		}
	}
	a.sweepOnce()
	waitFor(t, "the first batch to start", func() bool { return b.workerStarts() == maxWorkerStartsPerSweep })
	if n := b.workerStarts(); n != maxWorkerStartsPerSweep {
		t.Fatalf("workerStarts = %d in one sweep, want the per-sweep bound %d", n, maxWorkerStartsPerSweep)
	}
	a.sweepOnce() // the remainder comes up on the next tick
	waitFor(t, "the remainder to start", func() bool { return b.workerStarts() == maxWorkerStartsPerSweep+3 })
}

// A supervised restart must not exec while holding the activator's global
// mutex: ExecBackend.StartWorker materializes the build from the bucket when
// live/<host> is dangling, and a slow fetch held across a.mu stalls every
// request wake, Owns and Shutdown on the box.
func TestWorkerCrashRestartSpawnsOffTheActivatorLock(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.mu.Lock()
	b.workerStartDelay = 400 * time.Millisecond // a slow exec: a bucket fetch
	b.mu.Unlock()
	b.lastWorker().kill()

	start := time.Now()
	a.sweepOnce()
	if elapsed := time.Since(start); elapsed > 150*time.Millisecond {
		t.Fatalf("sweepOnce took %s — the sweeper is doing the exec itself, under a.mu", elapsed)
	}
	waitFor(t, "the restart to be in flight", func() bool { return a.entryRespawning("bot.x.internal") })

	// The lock must be free for everyone else while the spawn runs.
	start = time.Now()
	owns := a.Owns("bot.x.internal")
	if elapsed := time.Since(start); elapsed > 150*time.Millisecond {
		t.Fatalf("Owns blocked %s behind a respawn — the spawn is holding the global mutex", elapsed)
	}
	if !owns {
		t.Fatal("Owns lost the entry mid-restart")
	}
	waitFor(t, "the restart to land", func() bool { return b.workerStarts() == 2 })
}

// A respawn must never spawn a child after the lease has gone. Shutdown
// announces itself first, so a respawn still in flight abandons rather than
// starting a process that would write the tenant's DB with no single-writer
// claim behind it — and possibly outlive the agent as an orphan.
func TestShutdownStopsARespawnFromOutlivingTheLease(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.mu.Lock()
	b.workerStartDelay = 500 * time.Millisecond
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce() // starts the respawn
	waitFor(t, "the respawn to be in flight", func() bool { return a.entryRespawning("bot.x.internal") })

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	a.Shutdown(ctx)

	if a.entryCount() != 0 {
		t.Fatal("Shutdown returned with entries still live")
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d at shutdown-return — a child was spawned after the lease was released", n)
	}
	// …and none appears afterwards either: the goroutine must have abandoned,
	// not merely been outrun.
	time.Sleep(700 * time.Millisecond)
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d after shutdown — an orphan was spawned once the agent had gone", n)
	}
	for i, p := range b.workers {
		if !p.isStopped() {
			t.Fatalf("worker child %d was left running by Shutdown", i)
		}
	}
}

// THE invariant this task has been circling: when respawnDone is observed
// closed, no child spawned by that respawn is still running.
//
// The case that proves it is the raced one — a stop lands after the respawn
// has committed to spawning, so the child it produces must be SHED. If
// respawnDone closed before that shed finished, stop() would fall through and
// reach rep.Close — final sync plus lease release — with a live process still
// pointed at the same database. Narrowing that window is not closing it: the
// shed has to complete first.
func TestRespawnShedsARacedChildBeforeTheLeaseIsReleased(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	rep := b.lastRep()
	// The gate applies to children spawned from now on — i.e. the respawn's,
	// the one that will have to be shed. Its drain is held open so the
	// ordering is observable rather than a microsecond wide.
	gate := make(chan struct{})
	b.mu.Lock()
	b.workerStartDelay = 200 * time.Millisecond
	b.workerStopGate = gate
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the respawn's exec to begin", func() bool { return b.workerEnters() == 2 })

	// The stop lands mid-exec: past the pre-spawn check, so the child is
	// already coming and can only be shed.
	if !a.Bounce("bot.x.internal") {
		t.Fatal("Bounce found no awake worker to stop")
	}
	waitFor(t, "the raced child to be spawned", func() bool { return b.workerStarts() == 2 })

	// It is now draining, held by the gate. The lease must still be held.
	deadline := time.Now().Add(300 * time.Millisecond)
	for time.Now().Before(deadline) {
		if rep.isClosed() {
			t.Fatal("the lease was released while the respawn's raced child was still draining — a live process is pointed at a database we no longer hold")
		}
		time.Sleep(10 * time.Millisecond)
	}

	close(gate) // let the shed finish
	waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 })
	if n := rep.closeCount(); n != 1 {
		t.Fatalf("lease closes = %d, want exactly 1", n)
	}
	for i, p := range b.workers {
		if !p.isStopped() {
			t.Fatalf("worker child %d survived the stop", i)
		}
	}
}

// The same invariant, but against the SCHEDULING order rather than the exec:
// the stop goroutine arrives after the respawn has cleared its interlock and
// while the shed is still draining. Gating the wait on that flag would skip it
// here — the flag is cleared in the respawn's final critical section, before
// the shed — and the lease would go while a child was still alive.
//
// beginStopLocked flips the state under a.mu and then launches the stop, so in
// production the two race for the lock with nothing ordering them. This test
// arranges the losing order deliberately, from inside the package: it holds
// a.mu until the respawn's final critical section is queued behind it, condemns
// the entry, and releases — Go hands a mutex directly to a waiter that has
// waited past its starvation threshold, so the respawn runs first and the stop
// goroutine reads the entry second. No invariant is bent to produce it; only
// the schedule is pinned.
func TestStopScheduledAfterTheInterlockClearStillWaitsForTheShed(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	rep := b.lastRep()
	gate := make(chan struct{})
	b.mu.Lock()
	b.workerStartDelay = 150 * time.Millisecond
	b.workerStopGate = gate // held on the respawn's child: its drain blocks
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the respawn's exec to begin", func() bool { return b.workerEnters() == 2 })

	// Take the lock while the exec is still running, and hold it: the respawn
	// finishes and queues behind us for its final critical section. THEN
	// condemn the entry — so its `live` check will fail and it must shed —
	// and launch the stop, which queues behind the respawn. On release the
	// respawn goes first (clearing the interlock, entering its gated shed) and
	// the stop reads second: the order that used to skip the wait.
	a.mu.Lock()
	time.Sleep(300 * time.Millisecond) // exec completes; that goroutine queues on a.mu
	e := a.entries["bot.x.internal"]
	if e != nil {
		a.beginStopLocked(e, StopIdle)
	}
	a.mu.Unlock()
	if e == nil {
		t.Fatal("the worker entry vanished before the stop could be arranged")
	}
	waitFor(t, "the raced child to be spawned", func() bool { return b.workerStarts() == 2 })

	// The shed is now draining behind the gate. The lease must still be held:
	// releasing it here is releasing it under a live process.
	deadline := time.Now().Add(300 * time.Millisecond)
	for time.Now().Before(deadline) {
		if rep.isClosed() {
			t.Fatal("the lease was released while the respawn's raced child was still draining — the stop skipped its wait because the interlock had already been cleared")
		}
		time.Sleep(10 * time.Millisecond)
	}

	close(gate)
	waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 })
	if n := rep.closeCount(); n != 1 {
		t.Fatalf("lease closes = %d, want exactly 1", n)
	}
	for i, p := range b.workers {
		if !p.isStopped() {
			t.Fatalf("worker child %d survived the stop", i)
		}
	}
}

// Shutdown CANCELS an in-flight worker spawn rather than waiting it out. The
// wait is right everywhere else, but a shutdown answers to a deadline it does
// not set: systemd's TimeoutStopSec defaults to 90s and workerSpawnTimeout
// alone is 2 minutes, so blocking on one host's bucket materialize earns a
// SIGKILL that strands leases and sockets for EVERY host on the box.
func TestShutdownCancelsAnInFlightRespawnInsteadOfWaiting(t *testing.T) {
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	rep := b.lastRep()
	b.mu.Lock()
	b.workerStartDelay = 5 * time.Second // a materialize nobody should wait for
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	// Inside the exec: past the pre-spawn check, so only cancellation — not
	// the shutdown announcement — can end this promptly.
	waitFor(t, "the respawn's exec to begin", func() bool { return b.workerEnters() == 2 })

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	start := time.Now()
	a.Shutdown(ctx)
	if elapsed := time.Since(start); elapsed > 2*time.Second {
		t.Fatalf("Shutdown took %s — it waited out the spawn instead of cancelling it, which is how an agent gets SIGKILLed mid-shutdown", elapsed)
	}

	// (c) nothing left wedged, and no child running.
	if a.entryCount() != 0 {
		t.Fatal("Shutdown returned with an entry still live")
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d — a cancelled exec must produce no child", n)
	}
	for i, p := range b.workers {
		if !p.isStopped() {
			t.Fatalf("worker child %d was left running by Shutdown", i)
		}
	}
	// The lease: released exactly once, by stop()'s rep.Close — the respawn
	// never owned it, so cancelling one can neither skip nor double the
	// release.
	if n := rep.closeCount(); n != 1 {
		t.Fatalf("lease closes = %d, want exactly 1 on the cancelled-respawn shutdown path", n)
	}
	// releaseCount counts the lease coming FREE by either route — fakeRep.Close
	// after a wake got past Replicate, or a raw Lease.Release before it (see
	// fakeRep.b). The two are mutually exclusive per wake, so exactly 1 here
	// says the same release rep.closeCount() just counted and no second one
	// beside it: cancelling a respawn that never owned the lease can neither
	// skip the release nor double it.
	if n := b.releaseCount(); n != 1 {
		t.Fatalf("lease releases = %d, want exactly 1 — the lease was traded for replication at wake, so it comes free through rep.Close and nowhere else", n)
	}
	// And the abandoned goroutine really is gone: nothing appears later.
	time.Sleep(300 * time.Millisecond)
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d after shutdown — the cancelled respawn spawned anyway", n)
	}
}

// A stop that lands AFTER a respawn has committed to spawning must not release
// the lease under it: the child appearing moments later would write the
// tenant's DB with no single-writer claim behind it. Bounce is the case
// Shutdown's announcement does not cover — and, unlike a shutdown, it has no
// deadline, so here the stop WAITS the spawn out rather than cancelling it.
func TestStopWaitsOutAnInFlightRespawn(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.mu.Lock()
	b.workerStartDelay = 400 * time.Millisecond
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	// Inside the exec, i.e. PAST the respawn's own pre-spawn check: from here
	// a stop can only be raced, which is what the wait exists for.
	waitFor(t, "the respawn's exec to begin", func() bool { return b.workerEnters() == 2 })

	if !a.Bounce("bot.x.internal") {
		t.Fatal("Bounce found no awake worker to stop")
	}
	// The lease must still be held for as long as the exec is running. Watched
	// against the SPAWN's own progress (workerStarts rises only when
	// StartWorker returns), not against the entry: a stop that ignored the
	// respawn would delete the entry in the same breath as closing the lease,
	// leaving nothing to ask.
	deadline := time.Now().Add(300 * time.Millisecond)
	for time.Now().Before(deadline) {
		if b.lastRep().isClosed() && b.workerStarts() < 2 {
			t.Fatal("the lease was released while a respawn's exec was still running — the child appearing behind it would have no single-writer claim")
		}
		time.Sleep(10 * time.Millisecond)
	}

	waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 })
	for i, p := range b.workers {
		if !p.isStopped() {
			t.Fatalf("worker child %d survived the stop", i)
		}
	}
}

// --- always-on workers: converging onto a shipped build -----------------------

// shippingActivator wires the fake backend and the sweep's liveVer seam to ONE
// modelled live/<host> symlink, so a test can "ship" by moving it: a child
// execs whatever it points at now, and the sweep reads the same source.
func shippingActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeClock, *atomic.Value) {
	t.Helper()
	a, b, clk, _ := testActivator(t, idle)
	var ver atomic.Value
	ver.Store("v1")
	b.execVerFn = func() string { return ver.Load().(string) }
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	return a, b, clk, &ver
}

// shippingActivatorWithRegistry is shippingActivator with the registry handle
// kept — the version-clear-on-respawn tests need to read routes.version back,
// which shippingActivator's callers have never needed before.
func shippingActivatorWithRegistry(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeClock, *registry.Registry, *atomic.Value) {
	t.Helper()
	a, b, clk, reg := testActivator(t, idle)
	var ver atomic.Value
	ver.Store("v1")
	b.execVerFn = func() string { return ver.Load().(string) }
	a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
	return a, b, clk, reg, &ver
}

// A running child never re-reads live/<host>, and an always-on worker never
// sleeps, so without this it would serve the old binary forever while
// `carlos deploy` reported success. It converges as one MOVE — stop, respawn,
// same entry, same lease — because the ordinary cycle's second half ("a
// request wakes it onto the new build") does not exist for a worker.
func TestAlwaysOnWorkerRespawnsOntoANewBuild(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	if got := a.entrySpawnVersion("bot.x.internal"); got != "v1" {
		t.Fatalf("spawn version = %q, want v1", got)
	}
	first := b.lastWorker()

	ver.Store("v2") // a ship repoints the symlink under the running child
	a.sweepOnce()   // first sighting: starts the grace clock, never acts
	// entryRespawning, not a spawn count: convergeWorkerLocked sets the
	// interlock synchronously, under the same lock the sweep holds, so this
	// sees a respawn that has merely been STARTED — a count would race the
	// goroutine and pass whether or not the grace window was honoured.
	if a.entryRespawning("bot.x.internal") {
		t.Fatal("a respawn began on the drift's first sighting — the ConvergeGrace window must pass first")
	}
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d — the grace window must pass before anything is respawned", n)
	}
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()

	waitFor(t, "the respawn onto v2", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
	if n := b.workerStarts(); n != 2 {
		t.Fatalf("workerStarts = %d, want 2 (one respawn)", n)
	}
	if !first.isStopped() {
		t.Fatal("the old child was left running beside its replacement — a sidecar would double its work")
	}
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("the worker is not awake after converging — the respawn must land on the SAME entry")
	}
	if b.lastRep().isClosed() || b.releaseCount() != 0 {
		t.Fatal("converging respawned through a stop: the lease must not churn for an always-on worker")
	}
	// The new child holds the live token, so the agent socket still knows it.
	if tok := a.WorkerToken("bot.x.internal"); tok == "" || b.lastWorkerEnv()["CARLOS_SIDECAR_TOKEN"] != tok {
		t.Fatal("the respawned child was not handed the token the socket now expects")
	}
}

// The one-per-sweep budget still applies: a bad ship must not respawn every
// sidecar on the box at once.
func TestWorkerConvergeRespawnIsRateLimitedPerSweep(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	for _, host := range []string{"bot-a.x.internal", "bot-b.x.internal"} {
		if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
			t.Fatalf("WakeWorker %s: %v", host, err)
		}
	}
	ver.Store("v2")
	a.sweepOnce() // grace clocks start for both
	clk.advance(ConvergeGrace + time.Minute)

	a.sweepOnce()
	waitFor(t, "one respawn", func() bool { return b.workerStarts() == 3 })
	time.Sleep(100 * time.Millisecond) // give a second respawn every chance to appear
	if n := b.workerStarts(); n != 3 {
		t.Fatalf("workerStarts = %d after one sweep, want 3 — two workers, ONE respawn per tick", n)
	}

	// The next tick takes the other one.
	a.sweepOnce()
	waitFor(t, "the second respawn", func() bool { return b.workerStarts() == 4 })
}

// A HIBERNATING worker's convergence is unchanged: it is CYCLED (stopped, DB
// parked, lease released) and comes back on its schedule, exactly as before.
func TestHibernatingWorkerStillConvergenceCycles(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil { // Hibernate: true
		t.Fatalf("WakeWorker: %v", err)
	}
	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()

	waitFor(t, "the convergence cycle", func() bool { return a.entryCount() == 0 })
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d — a hibernating worker is cycled, not respawned in place", n)
	}
	if !b.lastRep().isClosed() {
		t.Fatal("the cycle did not release the lease")
	}
}

// A respawn stops the old child before spawning the new one, so for a moment
// the entry's proc reads as exited. A sweep landing in that window must do
// NOTHING: read as a crash it would start a second child, and the box would
// end up running two of a bot that must run once.
func TestWorkerConvergeRespawnIsNotMistakenForACrash(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.mu.Lock()
	b.workerStartDelay = 300 * time.Millisecond // hold the respawn's spawn open
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce() // starts the respawn

	waitFor(t, "the respawn to be in flight", func() bool { return a.entryRespawning("bot.x.internal") })
	a.sweepOnce() // a tick inside the window
	if n := b.workerStarts(); n != 1 {
		t.Fatalf("workerStarts = %d during a respawn — the sweep spawned a second child for a worker mid-converge", n)
	}

	waitFor(t, "the respawn to land", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
	if n := b.workerStarts(); n != 2 {
		t.Fatalf("workerStarts = %d after the respawn, want exactly 2", n)
	}
}

// A respawn whose spawn FAILS must not leave a hole: the dead child stays in
// place, and the ordinary crash-restart path picks it up on a later tick with
// its backoff. Convergence degrades into supervision.
func TestFailedConvergeRespawnFallsBackToCrashRestart(t *testing.T) {
	t.Parallel()
	a, b, clk, ver := shippingActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}
	b.mu.Lock()
	b.workerErr = errors.New("exec: live symlink gone")
	b.mu.Unlock()

	ver.Store("v2")
	a.sweepOnce()
	clk.advance(ConvergeGrace + time.Minute)
	a.sweepOnce()
	waitFor(t, "the failed respawn to finish", func() bool { return !a.entryRespawning("bot.x.internal") })
	if !a.entryAwake("bot.x.internal") {
		t.Fatal("a failed respawn tore the entry down — the crash restart would have nothing left to fix")
	}

	// The build comes back; the crash-restart path takes it from here.
	b.mu.Lock()
	b.workerErr = nil
	b.mu.Unlock()
	clk.advance(sidecarRestartBase + time.Second)
	a.sweepOnce()
	waitFor(t, "the crash restart to recover it", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
}

// --- /v1/idle and the always-on contract --------------------------------------

// "I have nothing to do" is not a request an always-on worker gets to make:
// the stop would delete its entry, leaving nothing — not even the crash
// restart — to bring it back before the agent restarts.
func TestSleepWorkerIgnoresAnAlwaysOnWorker(t *testing.T) {
	t.Parallel()
	a, b, _, _ := testActivator(t, time.Hour)
	if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
		t.Fatalf("WakeWorker: %v", err)
	}

	a.SleepWorker("bot.x.internal")

	if !a.entryAwake("bot.x.internal") {
		t.Fatal("an always-on worker obeyed /v1/idle — nothing would ever start it again")
	}
	if b.releaseCount() != 0 {
		t.Fatal("an always-on worker's lease was released on a /v1/idle it should have refused")
	}
}

// --- token minting ------------------------------------------------------------

// The agent supervises every route on the box. A wake that cannot mint a token
// must fail THAT wake — lease released, error reported — never take the
// process down with it.
func TestWorkerTokenMintFailureFailsWakeAndReleasesLease(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Hour)
	old := randRead
	randRead = func([]byte) (int, error) { return 0, errors.New("entropy pool exhausted") }
	defer func() { randRead = old }()

	err := a.WakeWorker(testWorkerRoute("bot.x.internal"))
	if err == nil {
		t.Fatal("a wake with no token reported success")
	}
	if !strings.Contains(err.Error(), "mint worker token") {
		t.Fatalf("wake error = %v, want it to name the mint failure", err)
	}
	if a.entryCount() != 0 {
		t.Fatal("the failed wake left an entry behind")
	}
	if n := b.releaseCount(); n != 1 {
		t.Fatalf("lease releases = %d, want exactly 1 — a mint failure is a wake failure like any other", n)
	}
	if n := b.workerStarts(); n != 0 {
		t.Fatalf("workerStarts = %d — nothing may be spawned without a token to authorize it", n)
	}
}

func TestConfigFromEnv(t *testing.T) {
	t.Setenv("CARLOS_HIBERNATE_BUCKET", "")
	if cfg, err := ConfigFromEnv(); err != nil || cfg != nil {
		t.Fatalf("unset bucket: got %+v, %v; want nil, nil (feature off)", cfg, err)
	}

	t.Setenv("CARLOS_HIBERNATE_BUCKET", "b")
	t.Setenv("CARLOS_HIBERNATE_REGION", "eu-west-1")
	t.Setenv("CARLOS_HIBERNATE_PREFIX", "/instances/")
	t.Setenv("CARLOS_HIBERNATE_IDLE", "90s")
	cfg, err := ConfigFromEnv()
	if err != nil {
		t.Fatalf("config: %v", err)
	}
	if cfg.Bucket != "b" || cfg.Region != "eu-west-1" || cfg.Prefix != "instances" || cfg.Idle != 90*time.Second {
		t.Fatalf("got %+v", cfg)
	}
	if got := cfg.hibernatePrefix("x.example.com"); got != "instances/x.example.com" {
		t.Fatalf("prefix: %q", got)
	}

	t.Setenv("CARLOS_HIBERNATE_IDLE", "")
	cfg, err = ConfigFromEnv()
	if err != nil || cfg.Idle != 15*time.Minute {
		t.Fatalf("default idle: %+v, %v", cfg, err)
	}

	t.Setenv("CARLOS_HIBERNATE_IDLE", "banana")
	if _, err := ConfigFromEnv(); err == nil {
		t.Fatal("bad idle duration should error, not default silently")
	}
}

// --- ExecBackend -------------------------------------------------------------

func testExecBackend(t *testing.T) *ExecBackend {
	t.Helper()
	return &ExecBackend{
		Cfg:     Config{Bucket: "test-bucket", Idle: time.Minute},
		OptDir:  t.TempDir(),
		DataDir: t.TempDir(),
	}
}

// stageLive materializes <OptDir>/live/<host> → <OptDir>/versions/<ver>/app,
// the layout the adopt pass maintains in production.
func stageLive(t *testing.T, b *ExecBackend, host, ver string) {
	t.Helper()
	vdir := filepath.Join(b.OptDir, "versions", ver)
	if err := os.MkdirAll(vdir, 0o755); err != nil {
		t.Fatal(err)
	}
	bin := filepath.Join(vdir, "app")
	if err := os.WriteFile(bin, []byte("#!/bin/sh\n"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	os.Remove(b.liveLink(host))
	if err := os.Symlink(bin, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}
}

// Already-materialized: the common case for every existing instance — the
// Materialize hook is never invoked.
func TestExecBackendLiveLinkAlreadyMaterialized(t *testing.T) {
	b := testExecBackend(t)
	stageLive(t, b, "existing.example.com", "v1")
	called := false
	b.Materialize = func(context.Context, string) error { called = true; return nil }
	if err := b.ensureLiveLink(context.Background(), "existing.example.com"); err != nil {
		t.Fatalf("ensureLiveLink: %v", err)
	}
	if called {
		t.Error("Materialize was called for an already-materialized host")
	}
}

// Missing symlink + Materialize configured: routed through the hook, and the
// hook's result is honored.
func TestExecBackendLiveLinkUsesMaterializeWhenConfigured(t *testing.T) {
	b := testExecBackend(t)
	var gotHost string
	b.Materialize = func(_ context.Context, host string) error {
		gotHost = host
		stageLive(t, b, host, "v1")
		return nil
	}
	if err := b.ensureLiveLink(context.Background(), "seeded.example.com"); err != nil {
		t.Fatalf("ensureLiveLink: %v", err)
	}
	if gotHost != "seeded.example.com" {
		t.Fatalf("Materialize called with %q, want seeded.example.com", gotHost)
	}
	if _, err := os.Stat(b.liveLink("seeded.example.com")); err != nil {
		t.Fatalf("live symlink was not created via Materialize: %v", err)
	}
}

// A Materialize failure must propagate — the exact live symptom found
// 2026-07-13 in Eleven (a 503 "instance failed to wake"), never silently
// swallowed into "looks fine, wakes anyway."
func TestExecBackendLiveLinkPropagatesMaterializeError(t *testing.T) {
	b := testExecBackend(t)
	wantErr := fmt.Errorf("controld materialize seeded.example.com: HTTP 502")
	b.Materialize = func(context.Context, string) error { return wantErr }
	err := b.ensureLiveLink(context.Background(), "seeded.example.com")
	if err == nil {
		t.Fatal("expected an error, got nil")
	}
	if !errors.Is(err, wantErr) {
		t.Fatalf("got %v, want it to wrap %v", err, wantErr)
	}
	if _, statErr := os.Stat(b.liveLink("seeded.example.com")); statErr == nil {
		t.Fatal("live symlink should not exist after a failed Materialize")
	}
}

// No hook and no symlink: an error, loudly. (Eleven self-staged its own
// binary here — instance and control binary were one program; a carlos app is
// a separate binary from the release store, so there is nothing to invent.)
func TestExecBackendLiveLinkMissingWithoutMaterializeFails(t *testing.T) {
	b := testExecBackend(t)
	if err := b.ensureLiveLink(context.Background(), "ghost.example.com"); err == nil {
		t.Fatal("expected an error for a missing live symlink with no Materialize hook")
	}
}

func TestExecBackendLiveVersion(t *testing.T) {
	b := testExecBackend(t)
	stageLive(t, b, "v.example.com", "abc1234")
	ver, err := b.LiveVersion("v.example.com")
	if err != nil || ver != "abc1234" {
		t.Fatalf("LiveVersion = %q, %v; want abc1234", ver, err)
	}
	// No symlink at all: "" with no error — version management not set up yet.
	ver, err = b.LiveVersion("none.example.com")
	if err != nil || ver != "" {
		t.Fatalf("LiveVersion(no link) = %q, %v; want \"\", nil", ver, err)
	}
}

// A migrated tenant's live link points at <opt>/versions/<ver>/<acct>/<app>,
// so reading "the parent directory" returns the ACCOUNT SQID as the version.
// ConvergeDecision compares that against the route's version and would then
// believe every migrated tenant is converged forever — a tenant that silently
// stops taking deploys.
func TestExecBackendLiveVersionOnAccountScopedPath(t *testing.T) {
	b := testExecBackend(t)
	stageLiveUnderAccount(t, b.OptDir, "kass.example.com", "kqmvexample", "abc1234", "kass")
	if ver, err := b.LiveVersion("kass.example.com"); err != nil || ver != "abc1234" {
		t.Fatalf("LiveVersion = %q, %v; want abc1234 (the version, not the account sqid)", ver, err)
	}
}

// stageLiveUnderAccount builds the account-scoped version-store layout and
// points a host's live symlink at it.
func stageLiveUnderAccount(t *testing.T, optDir, host, acct, ver, app string) {
	t.Helper()
	dir := filepath.Join(optDir, "versions", ver, acct)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, app), []byte("bin"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(optDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(filepath.Join(dir, app), filepath.Join(optDir, "live", host)); err != nil {
		t.Fatal(err)
	}
}

// childEnv strips every CARLOS_* var from the inherited environment (the
// edge's own config must never leak into a child) and then layers the env
// file on top — including that file's own CARLOS_* entries.
func TestChildEnvStripsCarlosVars(t *testing.T) {
	t.Setenv("CARLOS_HIBERNATE_BUCKET", "leak-me-not")
	t.Setenv("CARLOS_TEST_HARMLESS", "also-me")
	t.Setenv("UNRELATED_VAR", "keep")
	env := childEnv(map[string]string{"CARLOS_ADMIN_TOKEN": "tok", "FOO": "bar"})
	got := map[string]string{}
	for _, kv := range env {
		if k, v, ok := splitEnv(kv); ok {
			got[k] = v
		}
	}
	if _, leaked := got["CARLOS_HIBERNATE_BUCKET"]; leaked {
		t.Error("inherited CARLOS_HIBERNATE_BUCKET leaked into the child env")
	}
	if _, leaked := got["CARLOS_TEST_HARMLESS"]; leaked {
		t.Error("inherited CARLOS_TEST_HARMLESS leaked into the child env")
	}
	if got["UNRELATED_VAR"] != "keep" {
		t.Error("non-CARLOS inherited var was stripped")
	}
	if got["CARLOS_ADMIN_TOKEN"] != "tok" || got["FOO"] != "bar" {
		t.Errorf("env file vars not layered: %v", got)
	}
}

func splitEnv(kv string) (k, v string, ok bool) {
	for i := 0; i < len(kv); i++ {
		if kv[i] == '=' {
			return kv[:i], kv[i+1:], true
		}
	}
	return "", "", false
}

// mergeEnv layers the sync-owned <host>.config.env UNDER the instance's
// own <host>.env: the instance file wins every conflict, which is what
// makes CARLOS_ADMIN_TOKEN structurally unclobberable from the bucket.
func TestMergeEnvInstanceFileWins(t *testing.T) {
	base := map[string]string{"A": "bucket", "B": "bucket-only", "CARLOS_ADMIN_TOKEN": "forged"}
	overlay := map[string]string{"A": "instance", "CARLOS_ADMIN_TOKEN": "real"}
	got := mergeEnv(base, overlay)
	if got["A"] != "instance" || got["B"] != "bucket-only" || got["CARLOS_ADMIN_TOKEN"] != "real" {
		t.Fatalf("mergeEnv = %v, want instance values winning and bucket-only surviving", got)
	}
	// Inputs must not be mutated (withVar's same discipline).
	if base["A"] != "bucket" || len(overlay) != 2 {
		t.Fatal("mergeEnv mutated an input map")
	}
}

func TestMergeEnvNilInputs(t *testing.T) {
	if got := mergeEnv(nil, map[string]string{"A": "1"}); got["A"] != "1" {
		t.Fatalf("nil base: got %v", got)
	}
	if got := mergeEnv(map[string]string{"A": "1"}, nil); got["A"] != "1" {
		t.Fatalf("nil overlay: got %v", got)
	}
	if got := mergeEnv(nil, nil); len(got) != 0 {
		t.Fatalf("nil/nil: got %v", got)
	}
}

// The sleep-time next-due probe: GET /api/next-due over the instance's unix
// socket, authorized with the CARLOS_ADMIN_TOKEN from its env file.
func TestExecBackendNextDue(t *testing.T) {
	requireUnixSockets(t)
	b := testExecBackend(t)
	sock := tmpSock(t)
	ln, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatalf("listen unix: %v", err)
	}
	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/next-due" {
			http.NotFound(w, r)
			return
		}
		if r.Header.Get("Authorization") != "Bearer tok" {
			http.Error(w, "forbidden", http.StatusForbidden)
			return
		}
		io.WriteString(w, `{"due": 42}`)
	})}
	go srv.Serve(ln)
	t.Cleanup(func() { srv.Close() })

	host := "due.example.com"
	rt := registry.Route{Host: host, Kind: "instance", Socket: sock, Hibernate: true}

	// No env file at all: the probe fails (the state machine records 0).
	if _, err := b.NextDue(context.Background(), rt); err == nil {
		t.Fatal("NextDue with no env file should error")
	}
	// Env file without a token: same.
	envPath := filepath.Join(b.DataDir, host+".env")
	if err := os.WriteFile(envPath, []byte("# no token here\nFOO=bar\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := b.NextDue(context.Background(), rt); err == nil {
		t.Fatal("NextDue without CARLOS_ADMIN_TOKEN should error")
	}
	// With the token: the instance's answer comes back.
	if err := os.WriteFile(envPath, []byte("CARLOS_ADMIN_TOKEN=tok\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	due, err := b.NextDue(context.Background(), rt)
	if err != nil || due != 42 {
		t.Fatalf("NextDue = %d, %v; want 42", due, err)
	}
}

// TestExecBackendNextDueRefusesAFIFOEnvFile pins NextDue's nested dispatch:
// its env file lives on tenant-writable ground exactly like instanceEnv's
// does, so a FIFO planted there must fail the read fast rather than block
// the probe forever.
func TestExecBackendNextDueRefusesAFIFOEnvFile(t *testing.T) {
	b, rt := nestedBackend(t, &fakeControl{}, &fakeUnit{})
	instDir := provision.InstanceDir(b.DataDir, rt.Host)
	if err := os.MkdirAll(instDir, 0o755); err != nil {
		t.Fatalf("mkdir instance dir: %v", err)
	}
	envPath := filepath.Join(instDir, rt.Host+".env")
	if err := unix.Mkfifo(envPath, 0o600); err != nil {
		t.Fatalf("Mkfifo: %v", err)
	}
	done := make(chan error, 1)
	go func() {
		_, err := b.NextDue(context.Background(), rt)
		done <- err
	}()
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("NextDue against a FIFO env file succeeded, want an error")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("NextDue blocked on a FIFO env file instead of refusing it")
	}
}

// HasSidecar is derived from the filesystem — the /data/<host>-sidecar.env
// convention — no registry column to drift from what's actually staged.
func TestExecBackendHasSidecar(t *testing.T) {
	b := testExecBackend(t)
	if b.HasSidecar(testRoute("plain.example.com")) {
		t.Fatal("HasSidecar true with no env file staged")
	}
	if err := os.WriteFile(filepath.Join(b.DataDir, "bots.example.com-sidecar.env"), []byte("CARLOS_BASE=x\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	if !b.HasSidecar(testRoute("bots.example.com")) {
		t.Fatal("HasSidecar false with the sidecar env file staged")
	}
}

// StartWorker spawns a child that binds nothing — no socket, no port — so it
// must return the moment the exec succeeds, never waiting on waitHealthyCtx
// (which would dial unix://"" every 150ms for the full spawnHealthTimeout,
// 30s by default in this test, and still fail). The assertion is on elapsed
// wall time against a child that sleeps well past any reasonable spawn
// latency, specifically so a regression that reintroduces a health wait
// fails this test instead of merely running slow.
func TestExecBackendStartWorkerSpawnsWithoutAnyHealthProbe(t *testing.T) {
	b := testExecBackend(t)
	host := "bot.x.internal"
	stageSeedScript(t, b, host, "v1", "sleep 30\n")

	rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`,
		DBPath: filepath.Join(b.DataDir, "bot.db")}

	start := time.Now()
	p, _, err := b.StartWorker(context.Background(), rt, map[string]string{"CARLOS_SIDECAR_NAME": "bot"})
	if err != nil {
		t.Fatalf("StartWorker: %v", err)
	}
	defer p.Stop(context.Background())
	if d := time.Since(start); d > 5*time.Second {
		t.Fatalf("StartWorker took %v — it waited for a health probe that can never answer", d)
	}
	if p.Exited() {
		t.Fatal("worker exited immediately")
	}
}

// A route with no runnable argv has nothing to exec — refusing beats
// spawning the bare binary with no arguments (which for most binaries is
// either a no-op forever or an immediate usage-error crash loop).
func TestExecBackendStartWorkerRefusesEmptyArgs(t *testing.T) {
	b := testExecBackend(t)
	if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "b.x.internal", Kind: "sidecar"}, nil); err == nil {
		t.Fatal("a worker with no argv has nothing to run — refuse rather than exec the bare binary")
	}
}

// Malformed (not valid JSON) argv is refused exactly like empty argv, not
// silently coerced into "run nothing" or "run the bare binary".
func TestExecBackendStartWorkerRefusesMalformedArgs(t *testing.T) {
	b := testExecBackend(t)
	if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "b.x.internal", Kind: "sidecar", Args: "not json"}, nil); err == nil {
		t.Fatal("malformed args was silently accepted")
	}
}

// The env argument is merged OVER the route's own env layers and wins every
// conflict: it carries the per-wake sidecar token, which a tenant's own env
// file must never be able to override.
func TestExecBackendStartWorkerEnvWinsOverTenantFile(t *testing.T) {
	b := testExecBackend(t)
	host := "worker-env.x.internal"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	ownEnv := "CARLOS_SIDECAR_TOKEN=tenant-forged\nSHARED=instance\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte(ownEnv), 0o600); err != nil {
		t.Fatal(err)
	}

	rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`}
	p, _, err := b.StartWorker(context.Background(), rt, map[string]string{"CARLOS_SIDECAR_TOKEN": "agent-issued"})
	if err != nil {
		t.Fatalf("StartWorker: %v", err)
	}
	defer p.Stop(context.Background())

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got := child["CARLOS_SIDECAR_TOKEN"]; got != "agent-issued" {
		t.Fatalf("CARLOS_SIDECAR_TOKEN = %q, want %q — the agent-issued token must win over the tenant's own env file", got, "agent-issued")
	}
	if got := child["SHARED"]; got != "instance" {
		t.Fatalf("SHARED = %q, want %q — the route's own env layers must still flow through", got, "instance")
	}
}

// TestExecBackendStartWorkerEmitsExitEvent pins the fix for a gap Task 5's
// review carried into Task 15: Start's exit goroutine tells the member
// "instance exited" (warn on a crash, info on exit 0), but StartWorker's
// mirrored StartSidecar's silence instead. That mattered once sidecars got
// their own health words a member reads on the Instances tab (Crashed among
// them) — a crash with no event is a status word with no explanation. A
// nonzero exit must land a warn-level event naming the worker's host.
func TestExecBackendStartWorkerEmitsExitEvent(t *testing.T) {
	t.Parallel()
	b := testExecBackend(t)
	sink := &fakeSink{}
	b.Logs = sink
	host := "worker-exit.x.internal"
	stageSeedScript(t, b, host, "v1", "exit 7\n")

	rt := registry.Route{Host: host, Kind: "sidecar", App: "eleven", Args: `["sidecar","run"]`}
	p, _, err := b.StartWorker(context.Background(), rt, nil)
	if err != nil {
		t.Fatalf("StartWorker: %v", err)
	}
	waitFor(t, "worker exit", p.Exited)

	var found bool
	for _, e := range sink.events {
		if e.host == host && e.level == "warn" && strings.Contains(e.msg, "exited") {
			found = true
		}
	}
	if !found {
		t.Fatalf("no member-facing exit event for a crashed worker: %+v", sink.events)
	}
}

// stageRecordingLive stages host's live binary as a shell script that dumps
// its argv and its CARLOS_CONTROL_SOCKET to two files and exits — the only way
// to see what the child was ACTUALLY handed, argv and environment both.
// Returns the two paths.
func stageRecordingLive(t *testing.T, b *ExecBackend, host, ver string) (argvPath, envPath string) {
	t.Helper()
	dir := t.TempDir()
	argvPath = filepath.Join(dir, "argv")
	envPath = filepath.Join(dir, "env")
	script := "#!/bin/sh\n" +
		"printf '%s\\n' \"$@\" > '" + argvPath + "'\n" +
		"printf '%s\\n' \"${CARLOS_CONTROL_SOCKET-<unset>}\" > '" + envPath + "'\n"

	vdir := filepath.Join(b.OptDir, "versions", ver)
	if err := os.MkdirAll(vdir, 0o755); err != nil {
		t.Fatal(err)
	}
	bin := filepath.Join(vdir, "app")
	if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	os.Remove(b.liveLink(host))
	if err := os.Symlink(bin, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}
	return argvPath, envPath
}

// stageRecordingLiveFullEnv is stageRecordingLive's sibling: instead of
// picking out CARLOS_CONTROL_SOCKET alone, it dumps the child's ENTIRE
// real environment (`env`) to envPath. TestExecBackendStartMergesConfigEnvUnderOwnEnv
// needs the full dump — it isn't just checking that a var made it into the
// map Start builds, it's checking what the exec'd process itself actually
// saw.
func stageRecordingLiveFullEnv(t *testing.T, b *ExecBackend, host, ver string) (argvPath, envPath string) {
	t.Helper()
	dir := t.TempDir()
	argvPath = filepath.Join(dir, "argv")
	envPath = filepath.Join(dir, "env")
	script := "#!/bin/sh\n" +
		"printf '%s\\n' \"$@\" > '" + argvPath + ".tmp' && mv '" + argvPath + ".tmp' '" + argvPath + "'\n" +
		"env > '" + envPath + ".tmp' && mv '" + envPath + ".tmp' '" + envPath + "'\n"

	vdir := filepath.Join(b.OptDir, "versions", ver)
	if err := os.MkdirAll(vdir, 0o755); err != nil {
		t.Fatal(err)
	}
	bin := filepath.Join(vdir, "app")
	if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	os.Remove(b.liveLink(host))
	if err := os.Symlink(bin, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}
	return argvPath, envPath
}

// waitForFile polls until path exists (the child writes it concurrently with
// Start's health wait) and returns its contents.
func waitForFile(t *testing.T, path string, within time.Duration) string {
	t.Helper()
	deadline := time.Now().Add(within)
	for time.Now().Before(deadline) {
		if b, err := os.ReadFile(path); err == nil {
			return string(b)
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("%s never appeared within %s", path, within)
	return ""
}

// Start hands the spawned instance its control socket through the ENVIRONMENT
// (CARLOS_CONTROL_SOCKET), never as a CLI flag, and binds a listener on it.
// This is the WITH-CACHE case, where that listener also serves the purge
// verb; the nil-cache sibling below covers the socket existing anyway.
//
// The argv assertion is the regression guard: a --control-socket flag makes
// every app binary that predates this feature exit 2 at startup (Go's flag
// package with ExitOnError, which is what app binaries use), i.e. a
// fleet-wide outage the moment the agent upgrades. An unknown environment
// variable is simply ignored by such a binary.
//
// Deviation from the task brief's literal test: the brief asserted the
// control socket file still exists AFTER Start returns, even on the
// health-wait error the fake binary (which never binds rt.Socket) is expected
// to produce here. That's not actually true of the implementation the same
// brief specifies — Start's pre-existing health-wait-failure path calls
// p.Stop() before returning, and Step 3's own childProc.Stop change closes
// the purge listener there, which (per Go's net.Listener.Close on a unix
// socket the listener itself created) unlinks the socket file. So by the time
// Start returns with an error, the file is already gone — confirmed
// empirically, not a guess. This test instead polls for the socket WHILE
// Start's health-wait is still running (i.e. before its internal Stop cleanup
// fires), which is the invariant that's actually guaranteed and is what
// "Start wires up the purge listener" means in practice.
func TestExecBackendStartPassesControlSocketInEnvAndStartsPurgeListener(t *testing.T) {
	b := testExecBackend(t)
	b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
	argvPath, envPath := stageRecordingLive(t, b, "control-socket.example.com", "v1")

	rt := registry.Route{
		Host:   "control-socket.example.com",
		Kind:   "instance",
		Socket: tmpSock(t),
	}
	wantControlSocket := rt.Socket + ".control"

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	done := make(chan struct{})
	var proc Proc
	go func() {
		proc, _, _ = b.Start(ctx, rt) // health-wait error from the fake binary is expected and irrelevant here
		close(done)
	}()

	found := false
	deadline := time.Now().Add(1500 * time.Millisecond)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(wantControlSocket); err == nil {
			found = true
			break
		}
		time.Sleep(10 * time.Millisecond)
	}

	gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
	gotEnv := strings.TrimSpace(waitForFile(t, envPath, time.Second))

	// Everything above was seen while the health wait ran; the fake never
	// binds, so end the wait rather than sit out its timeout.
	cancel()
	<-done
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	wantArgv := []string{"--socket", rt.Socket, "--db", filepath.Join(b.DataDir, rt.Host+".db")}
	if !slices.Equal(gotArgv, wantArgv) {
		t.Fatalf("child argv = %v, want %v — a flag the app binary doesn't parse makes it exit 2 at startup", gotArgv, wantArgv)
	}
	if gotEnv != wantControlSocket {
		t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q", gotEnv, wantControlSocket)
	}
	if !found {
		t.Fatalf("expected control socket %s to exist while Start's health-wait ran", wantControlSocket)
	}
}

// The control channel is no longer the purge channel. It is bound for EVERY
// exec-backed child, cache or no cache (spec 2026-08-23-scheduled-work-design
// §7.2 — the scheduled-work verbs are served on it, and a box with no edge
// cache still runs scheduled work), so a nil ExecBackend.Cache no longer means
// "no control socket". What it does still mean is no CARLOS_PURGE: nothing on
// this box can purge anything, and the rule that a variable naming a channel
// must not be set where that channel does nothing is exactly what makes the
// two names diverge here.
//
// The same spawn now also mints the instance's CARLOS_ADMIN_TOKEN before it
// reads the env, so the child is handed the same secret the agent presents
// back to it (NextDue, the tick) and the control socket checks bearers
// against — nothing minted it before this, which is why the next-due probe
// silently failed for every console-created instance.
func TestExecBackendStartWithNilCacheStillBindsControlSocket(t *testing.T) {
	b := testExecBackend(t) // b.Cache left nil
	host := "no-cache.example.com"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: tmpSock(t),
	}
	wantControlSocket := rt.Socket + ".control"

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	// Same shape as the cached case above: poll for the socket WHILE Start's
	// health-wait is still running, because the fake binary never binds
	// rt.Socket and Start's failure path Stops the child, which unlinks the
	// control socket before Start returns.
	done := make(chan struct{})
	var proc Proc
	go func() {
		proc, _, _ = b.Start(ctx, rt) // health-wait error from the fake binary is expected here
		close(done)
	}()

	found := false
	deadline := time.Now().Add(1500 * time.Millisecond)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(wantControlSocket); err == nil {
			found = true
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}

	// Everything above was seen while the health wait ran; the fake never
	// binds, so end the wait rather than sit out its timeout.
	cancel()
	<-done
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	if !found {
		t.Fatalf("expected control socket %s while Start's health-wait ran — it is bound for every exec child now, cache or not", wantControlSocket)
	}
	if got := child["CARLOS_CONTROL_SOCKET"]; got != wantControlSocket {
		t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q", got, wantControlSocket)
	}
	if got, ok := child["CARLOS_PURGE"]; ok {
		t.Fatalf("child CARLOS_PURGE = %q, want it unset when there is no cache to purge", got)
	}

	minted, err := readEnvFile(filepath.Join(b.DataDir, host+".env"))
	if err != nil {
		t.Fatalf("read the instance env file the spawn should have minted into: %v", err)
	}
	if len(minted["CARLOS_ADMIN_TOKEN"]) != 64 {
		t.Fatalf("minted CARLOS_ADMIN_TOKEN = %q, want 64 hex chars", minted["CARLOS_ADMIN_TOKEN"])
	}
	if got := child["CARLOS_ADMIN_TOKEN"]; got != minted["CARLOS_ADMIN_TOKEN"] {
		t.Fatalf("child CARLOS_ADMIN_TOKEN = %q, want the minted %q — the child must hold the same secret the agent presents back to it", got, minted["CARLOS_ADMIN_TOKEN"])
	}
}

// fakeTimerStore is the registry subset schedule.Mount needs, stubbed: these
// tests are about the WIRING (are the verbs mounted, is the bearer the
// instance's minted token), not about timer semantics — internal/schedule
// owns those.
type fakeTimerStore struct{ rows []registry.Timer }

func (f *fakeTimerStore) UpsertTimer(t registry.Timer) error  { f.rows = append(f.rows, t); return nil }
func (f *fakeTimerStore) DeleteTimer(host, name string) error { return nil }
func (f *fakeTimerStore) GetTimer(host, name string) (*registry.Timer, error) {
	return nil, os.ErrNotExist
}
func (f *fakeTimerStore) TimersFor(host string) ([]registry.Timer, error) { return f.rows, nil }
func (f *fakeTimerStore) CountTimers(host, kind string) (int, error)      { return len(f.rows), nil }

// startAndReachControlSocket spawns host through the real Start path and hands
// the test the control socket's path plus the token that spawn minted, while
// Start's health-wait is STILL RUNNING — the fake binary never binds rt.Socket,
// so once Start returns its failure path has already Stopped the child and
// unlinked the control socket. Returns a stop func the caller must defer.
func startAndReachControlSocket(t *testing.T, b *ExecBackend, host string) (controlSocket, token string, stop func()) {
	t.Helper()
	stageRecordingLiveFullEnv(t, b, host, "v1")
	rt := registry.Route{Host: host, Kind: "instance", Socket: tmpSock(t)}
	controlSocket = rt.Socket + ".control"

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	done := make(chan struct{})
	var proc Proc
	go func() {
		proc, _, _ = b.Start(ctx, rt)
		close(done)
	}()

	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(controlSocket); err == nil {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	env, err := readEnvFile(filepath.Join(b.DataDir, host+".env"))
	if err != nil {
		cancel()
		<-done
		t.Fatalf("read minted env file: %v", err)
	}
	return controlSocket, env["CARLOS_ADMIN_TOKEN"], func() {
		cancel()
		<-done
		if proc != nil {
			proc.Stop(context.Background())
		}
	}
}

// The whole point of the generalised control channel: with ExecBackend.Timers
// set, the scheduled-work verbs are served on the spawned instance's control
// socket, gated on the token THIS spawn minted into <host>.env. The wrong
// bearer is a 401 — which keeps out other UNIX users on the box and any
// unauthenticated dialer, and is all this test claims. It does NOT close the
// same-uid-sibling gap purge.go documents: a sibling shares the agent's uid,
// so it can read another instance's 0600 env file and present that token.
// Per-instance uids are the fix; see purge.go for why the residual is bounded.
func TestExecBackendStartMountsScheduleVerbsOnControlSocket(t *testing.T) {
	b := testExecBackend(t) // b.Cache left nil — the schedule verbs do not need one
	b.Timers = &fakeTimerStore{}
	ctl, token, stop := startAndReachControlSocket(t, b, "timers.example.com")
	defer stop()

	if len(token) != 64 {
		t.Fatalf("minted CARLOS_ADMIN_TOKEN = %q, want 64 hex chars", token)
	}

	get := func(bearer string) int {
		req, err := http.NewRequest("GET", "http://carlos/schedule", nil)
		if err != nil {
			t.Fatal(err)
		}
		if bearer != "" {
			req.Header.Set("Authorization", "Bearer "+bearer)
		}
		resp, err := socketClient(ctl).Do(req)
		if err != nil {
			t.Fatalf("dial control socket %s: %v", ctl, err)
		}
		defer resp.Body.Close()
		return resp.StatusCode
	}

	if got := get(token); got != http.StatusOK {
		t.Fatalf("GET /schedule with the minted token = %d, want 200 — the verbs are not mounted", got)
	}
	if got := get("not-the-token"); got != http.StatusUnauthorized {
		t.Fatalf("GET /schedule with a wrong bearer = %d, want 401 — a same-uid sibling could reach this socket", got)
	}
	if got := get(""); got != http.StatusUnauthorized {
		t.Fatalf("GET /schedule with no bearer = %d, want 401", got)
	}
}

// Timers is nil-tolerant, and honestly so: the socket is still bound (the
// channel is per-child now, not per-feature) but nothing answers /schedule.
// A 404 is the truth for an agent that predates the feature; a socket that
// wasn't there at all would make the child's CARLOS_CONTROL_SOCKET a lie.
func TestExecBackendStartWithNilTimersMountsNoScheduleVerbs(t *testing.T) {
	b := testExecBackend(t) // b.Timers and b.Cache both nil
	ctl, token, stop := startAndReachControlSocket(t, b, "no-timers.example.com")
	defer stop()

	req, err := http.NewRequest("GET", "http://carlos/schedule", nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Authorization", "Bearer "+token)
	resp, err := socketClient(ctl).Do(req)
	if err != nil {
		t.Fatalf("dial control socket %s: %v — it must exist even with no verbs mounted", ctl, err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("GET /schedule with nil Timers = %d, want 404", resp.StatusCode)
	}
}

// TestExecBackendStartMergesConfigEnvUnderOwnEnv is the load-bearing
// merge-precedence regression: unlike a unit test of mergeEnv in isolation,
// this goes through the real ExecBackend.Start spawn path and inspects the
// spawned process's ACTUAL environment (via `env`, dumped by the recorder
// script) — the only way to see what the child was really handed.
//
// It stages both <host>.config.env (sync-owned, bucket-derived —
// CARLOS_ADMIN_TOKEN=forged, SHARED=bucket, BUCKET_ONLY=1) and <host>.env
// (the instance's own file — CARLOS_ADMIN_TOKEN=real, SHARED=instance) and
// asserts the child saw the instance's own values winning the conflicts
// while the bucket-only var still flowed through. This must fail if
// Start's `mergeEnv(cfgEnv, ownEnv)` call is ever transposed to
// `mergeEnv(ownEnv, cfgEnv)`: that would let bucket-owned config forge
// CARLOS_ADMIN_TOKEN, a secret held only in the instance's own file.
// Verified empirically by transposing the call and watching this test
// fail, then restoring it (see the fix report for the RED output).
func TestExecBackendStartMergesConfigEnvUnderOwnEnv(t *testing.T) {
	b := testExecBackend(t)
	host := "merge-precedence.example.com"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	cfgEnv := "CARLOS_ADMIN_TOKEN=forged\nSHARED=bucket\nBUCKET_ONLY=1\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte(cfgEnv), 0o600); err != nil {
		t.Fatal(err)
	}
	ownEnv := "CARLOS_ADMIN_TOKEN=real\nSHARED=instance\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte(ownEnv), 0o600); err != nil {
		t.Fatal(err)
	}

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: filepath.Join(t.TempDir(), "instance.sock"),
	}

	ctx, cancel := recordedCtx(2*time.Second, envPath)
	defer cancel()
	proc, _, _ := b.Start(ctx, rt) // health-wait error from the fake binary is expected and irrelevant here
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got := child["CARLOS_ADMIN_TOKEN"]; got != "real" {
		t.Fatalf("child CARLOS_ADMIN_TOKEN = %q, want %q — the instance's own env must win over bucket config", got, "real")
	}
	if got := child["SHARED"]; got != "instance" {
		t.Fatalf("child SHARED = %q, want %q — the instance's own env must win over bucket config", got, "instance")
	}
	if got := child["BUCKET_ONLY"]; got != "1" {
		t.Fatalf("child BUCKET_ONLY = %q, want %q — bucket config must still flow through when not overridden", got, "1")
	}
}

// CARLOS_PURGE=header is the advisory half of the response-header purge
// channel (internal/edge/purge_header.go): it tells the app this box's edge
// reads Carlos-Purge, so the app can log or branch on it. It is ADDITIVE —
// CARLOS_CONTROL_SOCKET must still be there beside it, because the header is
// an additional channel and not a replacement — and it must never be a CLI
// flag, for the same flag.ExitOnError reason the control socket isn't.
//
// The gate matches CARLOS_CONTROL_SOCKET's (b.Cache != nil, i.e. "there is
// actually a cache to purge"), which is the same no-inert-feature discipline:
// a variable naming a channel must not be set on a box where that channel
// does nothing.
func TestExecBackendStartAdvertisesHeaderPurgeChannel(t *testing.T) {
	b := testExecBackend(t)
	b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
	host := "purge-header.example.com"
	argvPath, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: tmpSock(t),
	}

	ctx, cancel := recordedCtx(2*time.Second, envPath)
	defer cancel()
	proc, _, _ := b.Start(ctx, rt) // health-wait error from the fake binary is expected and irrelevant here
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got := child["CARLOS_PURGE"]; got != "header" {
		t.Fatalf("child CARLOS_PURGE = %q, want %q", got, "header")
	}
	if got := child["CARLOS_CONTROL_SOCKET"]; got != rt.Socket+".control" {
		t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q — the header channel is ADDITIONAL, it does not replace the socket", got, rt.Socket+".control")
	}
	gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
	wantArgv := []string{"--socket", rt.Socket, "--db", filepath.Join(b.DataDir, rt.Host+".db")}
	if !slices.Equal(gotArgv, wantArgv) {
		t.Fatalf("child argv = %v, want %v — a flag the app binary doesn't parse makes it exit 2 at startup", gotArgv, wantArgv)
	}
}

// The advisory var is gated on there being a cache at all: with none, there
// is no purge channel to advertise and setting the var would be a lie.
func TestExecBackendStartWithNilCacheSetsNoPurgeVar(t *testing.T) {
	b := testExecBackend(t) // b.Cache left nil
	host := "no-cache-purge.example.com"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: filepath.Join(t.TempDir(), "instance.sock"),
	}

	ctx, cancel := recordedCtx(2*time.Second, envPath)
	defer cancel()
	proc, _, _ := b.Start(ctx, rt)
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, _, ok := strings.Cut(line, "="); ok && k == "CARLOS_PURGE" {
			t.Fatalf("child saw %q with no cache configured", line)
		}
	}
}

// TestExecBackendStartTenantCannotClaimAbsentPurgeChannel closes the hole the
// gate alone left open. fileEnv is TENANT-supplied — <host>.config.env arrives
// from the bucket via `carlos env`, <host>.env is the instance's own file — so
// gating only on "does the agent SET the var" left a tenant free to set it
// itself. On a box with no cache the child would then be told a purge channel
// exists when none does, and an app that trusts CARLOS_PURGE would skip its
// own invalidation and serve stale content indefinitely. CARLOS_PURGE is
// therefore REMOVED here.
//
// CARLOS_CONTROL_SOCKET is the same hole and a worse one — a tenant-chosen
// socket path is a name the agent never vouched for — but its channel now
// exists on this box (bound for every exec child, §7.2), so the fix is
// OVERRIDE rather than removal: the child must be pointed at the socket the
// agent actually bound, never at /tmp/attacker.sock.
func TestExecBackendStartTenantCannotClaimAbsentPurgeChannel(t *testing.T) {
	b := testExecBackend(t) // b.Cache left nil — no purge channel on this box
	host := "liar.example.com"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	// The tenant declares both channels in its own config.
	tenantEnv := "CARLOS_PURGE=header\nCARLOS_CONTROL_SOCKET=/tmp/attacker.sock\nKEPT=1\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte(tenantEnv), 0o600); err != nil {
		t.Fatal(err)
	}

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: tmpSock(t),
	}

	ctx, cancel := recordedCtx(2*time.Second, envPath)
	defer cancel()
	proc, _, _ := b.Start(ctx, rt)
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got, ok := child["CARLOS_PURGE"]; ok {
		t.Fatalf("tenant-supplied CARLOS_PURGE=%q survived onto a box with no cache", got)
	}
	if want := rt.Socket + ".control"; child["CARLOS_CONTROL_SOCKET"] != want {
		t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q — a tenant must not redirect the control socket, cache or no cache",
			child["CARLOS_CONTROL_SOCKET"], want)
	}
	// Only the agent-owned names are stripped; ordinary tenant config is
	// untouched.
	if got := child["KEPT"]; got != "1" {
		t.Fatalf("child KEPT = %q, want %q — only the agent-owned vars may be removed", got, "1")
	}
}

// The other branch: with a cache, the agent's values must OVERRIDE whatever
// the tenant supplied, not merely be absent-or-present. A tenant that writes
// CARLOS_CONTROL_SOCKET=/tmp/attacker.sock must not have the child pointed at
// a socket the agent did not create.
func TestExecBackendStartOverridesTenantSuppliedPurgeVars(t *testing.T) {
	b := testExecBackend(t)
	b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
	host := "override.example.com"
	_, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	tenantEnv := "CARLOS_PURGE=socket-only\nCARLOS_CONTROL_SOCKET=/tmp/attacker.sock\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte(tenantEnv), 0o600); err != nil {
		t.Fatal(err)
	}

	rt := registry.Route{
		Host:   host,
		Kind:   "instance",
		Socket: tmpSock(t),
	}

	ctx, cancel := recordedCtx(2*time.Second, envPath)
	defer cancel()
	proc, _, _ := b.Start(ctx, rt)
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got := child["CARLOS_PURGE"]; got != "header" {
		t.Fatalf("child CARLOS_PURGE = %q, want %q — the agent's value must win over the tenant's", got, "header")
	}
	if want := rt.Socket + ".control"; child["CARLOS_CONTROL_SOCKET"] != want {
		t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q — a tenant must not redirect the control socket",
			child["CARLOS_CONTROL_SOCKET"], want)
	}
}

// Spawning a fresh process for a host is a deploy boundary: the code behind
// the host just changed, so every response cached from the old process is
// stale. Start drops them all before the new process serves anything —
// without this, a redeploy keeps serving old HTML until the TTL expires.
func TestExecBackendStartPurgesHostCacheOnSpawn(t *testing.T) {
	b := testExecBackend(t)
	cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
	b.Cache = cache
	stageLive(t, b, "deploy.example.com", "v1")

	stale := edge.CacheKey{Host: "deploy.example.com", Method: "GET", Path: "/event"}
	otherHost := edge.CacheKey{Host: "bystander.example.com", Method: "GET", Path: "/event"}
	for _, k := range []edge.CacheKey{stale, otherHost} {
		cache.Put(k, "", &edge.CachedResponse{
			StatusCode: 200, Header: http.Header{}, Body: []byte("old html"),
			StoredAt: time.Now(), TTL: time.Minute,
		})
	}

	// GetFor is the cache's request-time lookup (the by-VaryValue one is
	// package-internal); both entries were stored from responses with no Vary,
	// so any request for the URL reaches them.
	cached := func(k edge.CacheKey) bool {
		_, ok := cache.GetFor(k, httptest.NewRequest("GET", "http://"+k.Host+k.Path, nil))
		return ok
	}
	// The purge comes before the health wait, which the fake never answers:
	// once it has happened that wait has nothing left to show.
	ctx, cancel := ctxUntil(2*time.Second, func() bool { return !cached(stale) })
	defer cancel()
	rt := registry.Route{
		Host:   "deploy.example.com",
		Kind:   "instance",
		Socket: filepath.Join(t.TempDir(), "instance.sock"),
	}
	proc, _, _ := b.Start(ctx, rt) // health-wait failure is expected; the spawn still happened
	if proc != nil {
		defer proc.Stop(context.Background())
	}

	if cached(stale) {
		t.Fatal("spawning a fresh process for a host must drop that host's cached responses")
	}
	if !cached(otherHost) {
		t.Fatal("another host's cached responses must survive a spawn")
	}
}

// --- ExecBackend: fresh detection + seed --------------------------------------

// stageSeedScript stages host's live binary as a shell script with a chosen
// body — the seed tests need exits and stderr the recording scripts above
// don't produce. Returns nothing: each test knows the files its body writes.
func stageSeedScript(t *testing.T, b *ExecBackend, host, ver, body string) {
	t.Helper()
	vdir := filepath.Join(b.OptDir, "versions", ver)
	if err := os.MkdirAll(vdir, 0o755); err != nil {
		t.Fatal(err)
	}
	bin := filepath.Join(vdir, "app")
	if err := os.WriteFile(bin, []byte("#!/bin/sh\n"+body), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	os.Remove(b.liveLink(host))
	if err := os.Symlink(bin, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}
}

// The warm cache — the local DB still on disk from the last sleep — is never a
// first boot, and the bucket isn't even consulted.
func TestExecBackendRestoreWarmCacheIsNotFresh(t *testing.T) {
	b := testExecBackend(t)
	host := "warm.example.com"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".db"), []byte("db"), 0o600); err != nil {
		t.Fatal(err)
	}
	called := false
	swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
		called = true
		return false, nil
	})

	fresh, err := b.Restore(context.Background(), registry.Route{Host: host, Kind: "instance"})
	if err != nil {
		t.Fatalf("Restore: %v", err)
	}
	if fresh {
		t.Fatal("warm cache reported fresh — the seed would run over live data")
	}
	if called {
		t.Fatal("warm cache consulted the bucket")
	}
}

// DB absent but a replica exists: the data came back from S3, so this is a
// restore, not a first boot.
func TestExecBackendRestoreFromReplicaIsNotFresh(t *testing.T) {
	b := testExecBackend(t)
	swapRestoreReplica(t, func(_ context.Context, dbPath, _, _, _ string) (bool, error) {
		writeUsableDB(t, dbPath)
		return true, nil
	})

	fresh, err := b.Restore(context.Background(), registry.Route{Host: "restored.example.com", Kind: "instance"})
	if err != nil {
		t.Fatalf("Restore: %v", err)
	}
	if fresh {
		t.Fatal("a restored instance reported fresh — a seed here would overwrite recovered data")
	}
}

// A replica that comes back DAMAGED must fail the wake, not boot. Two things
// are on the line: `fresh` must stay false (a fresh verdict runs the route's
// seed, writing an empty database over recoverable data), and the bad file
// must not be left where the next wake's warm-cache branch will serve it
// unchecked.
func TestExecBackendRestoreCorruptReplicaFailsAndQuarantines(t *testing.T) {
	b := testExecBackend(t)
	host := "damaged.example.com"
	dbPath := filepath.Join(b.DataDir, host+".db")

	swapRestoreReplica(t, func(_ context.Context, out, _, _, _ string) (bool, error) {
		if err := os.WriteFile(out, []byte("this is not a sqlite database at all"), 0o600); err != nil {
			t.Fatal(err)
		}
		return true, nil
	})

	fresh, err := b.Restore(context.Background(), registry.Route{Host: host, Kind: "instance"})
	if err == nil {
		t.Fatal("Restore accepted a corrupt replica")
	}
	if fresh {
		t.Fatal("a corrupt restore reported fresh — the seed would run over recoverable data")
	}
	if _, statErr := os.Stat(dbPath); statErr == nil {
		t.Fatal("the corrupt file was left at the live path — the next wake would serve it as a warm cache")
	}
	if _, statErr := os.Stat(dbPath + quarantineSuffix); statErr != nil {
		t.Fatalf("no quarantined copy for the operator to look at: %v", statErr)
	}

	// And the next wake must genuinely re-attempt the restore rather than
	// short-circuit on leftover state.
	called := false
	swapRestoreReplica(t, func(_ context.Context, out, _, _, _ string) (bool, error) {
		called = true
		writeUsableDB(t, out)
		return true, nil
	})
	fresh, err = b.Restore(context.Background(), registry.Route{Host: host, Kind: "instance"})
	if err != nil {
		t.Fatalf("second Restore: %v", err)
	}
	if !called {
		t.Fatal("the second wake never consulted the bucket")
	}
	if fresh {
		t.Fatal("the recovered instance reported fresh")
	}
}

// writeUsableDB creates the kind of file a healthy restore produces: a real
// SQLite database with a schema, which is what CheckDB demands.
func writeUsableDB(t *testing.T, path string) {
	t.Helper()
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if _, err := db.Exec("CREATE TABLE t (id INTEGER PRIMARY KEY); INSERT INTO t VALUES (1)"); err != nil {
		t.Fatal(err)
	}
}

// DB absent AND no replica in the bucket: the one true first boot.
func TestExecBackendRestoreNoReplicaIsFresh(t *testing.T) {
	b := testExecBackend(t)
	swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
		return false, nil
	})

	fresh, err := b.Restore(context.Background(), registry.Route{Host: "brand-new.example.com", Kind: "instance"})
	if err != nil {
		t.Fatalf("Restore: %v", err)
	}
	if !fresh {
		t.Fatal("a never-replicated host did not report fresh — it would never be seeded")
	}
}

// A restore ERROR is not a first boot: "we couldn't tell" must never be
// reported as "there was nothing there".
func TestExecBackendRestoreErrorIsNotFresh(t *testing.T) {
	b := testExecBackend(t)
	swapRestoreReplica(t, func(context.Context, string, string, string, string) (bool, error) {
		return false, errors.New("s3: connection refused")
	})

	fresh, err := b.Restore(context.Background(), registry.Route{Host: "unreachable.example.com", Kind: "instance"})
	if err == nil {
		t.Fatal("Restore swallowed the replica error")
	}
	if fresh {
		t.Fatal("a failed restore reported fresh")
	}
}

// swapRestoreReplica points the replica-restore seam at a stub for one test.
func swapRestoreReplica(t *testing.T, fn func(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error)) {
	t.Helper()
	prev := restoreReplica
	restoreReplica = fn
	t.Cleanup(func() { restoreReplica = prev })
}

// The seed contract, end to end through a real exec: the app's declared argv
// followed by --db <path>, under the instance's own merged env layers, with
// the agent's CARLOS_* vars scrubbed and the purge/control vars (Start's, not
// the seed's) absent.
func TestExecBackendSeedRunsDeclaredArgvWithDBPath(t *testing.T) {
	b := testExecBackend(t)
	b.Cache = edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
	host := "seeded.example.com"
	argvPath, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")

	cfgEnv := "SHARED=bucket\nBUCKET_ONLY=1\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".config.env"), []byte(cfgEnv), 0o600); err != nil {
		t.Fatal(err)
	}
	// The tenant's own env file claims the purge channel. It is not the
	// tenant's to claim in either direction — those two names are
	// agent-owned (see withoutVars), and a seed told a control socket
	// exists at a path the agent never vouched for is exactly the leak the
	// spawn path's else-branch closes.
	ownEnv := "SHARED=instance\nCARLOS_ADMIN_TOKEN=real\nCARLOS_PURGE=header\nCARLOS_CONTROL_SOCKET=/tmp/tenant-chosen.sock\n"
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte(ownEnv), 0o600); err != nil {
		t.Fatal(err)
	}
	t.Setenv("CARLOS_HIBERNATE_BUCKET", "agent-only")

	rt := registry.Route{Host: host, Kind: "instance", Seed: `["seed","--fixtures"]`}
	if err := b.Seed(context.Background(), rt); err != nil {
		t.Fatalf("Seed: %v", err)
	}

	gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
	wantArgv := []string{"seed", "--fixtures", "--db", filepath.Join(b.DataDir, host+".db")}
	if !slices.Equal(gotArgv, wantArgv) {
		t.Fatalf("seed argv = %v, want %v", gotArgv, wantArgv)
	}

	child := map[string]string{}
	for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
		if k, v, ok := strings.Cut(line, "="); ok {
			child[k] = v
		}
	}
	if got := child["SHARED"]; got != "instance" {
		t.Fatalf("seed SHARED = %q, want %q — the instance's own env must win over bucket config", got, "instance")
	}
	if got := child["BUCKET_ONLY"]; got != "1" {
		t.Fatalf("seed BUCKET_ONLY = %q, want %q — bucket config must flow through", got, "1")
	}
	if got := child["CARLOS_ADMIN_TOKEN"]; got != "real" {
		t.Fatalf("seed CARLOS_ADMIN_TOKEN = %q, want %q — the seed gets the instance's own env file", got, "real")
	}
	if got := child["CARLOS_HOST"]; got != host {
		t.Fatalf("seed CARLOS_HOST = %q, want %q (#256: the seed is told its host like the instance is)", got, host)
	}
	if _, ok := child["CARLOS_HIBERNATE_BUCKET"]; ok {
		t.Fatal("the agent's own CARLOS_* config leaked into the seed process")
	}
	if _, ok := child["CARLOS_CONTROL_SOCKET"]; ok {
		t.Fatal("seed was handed a control socket — the purge channel belongs to the serving process, not a one-shot seed")
	}
	if _, ok := child["CARLOS_PURGE"]; ok {
		t.Fatal("seed was told a purge channel exists — that layering belongs to Start")
	}
}

// The route's own DB path wins over the <DataDir>/<host>.db default, exactly
// as it does for the spawn.
func TestExecBackendSeedHonoursRouteDBPath(t *testing.T) {
	b := testExecBackend(t)
	host := "custom-db.example.com"
	argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")
	dbPath := filepath.Join(t.TempDir(), "elsewhere.db")

	if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", DBPath: dbPath, Seed: `["seed"]`}); err != nil {
		t.Fatalf("Seed: %v", err)
	}
	gotArgv := strings.Fields(waitForFile(t, argvPath, time.Second))
	wantArgv := []string{"seed", "--db", dbPath}
	if !slices.Equal(gotArgv, wantArgv) {
		t.Fatalf("seed argv = %v, want %v", gotArgv, wantArgv)
	}
}

// A seed that exits non-zero fails, naming the host, the exit state, and what
// it printed — the wake dies with it (see doWake), so this error is the whole
// diagnosis an operator gets.
func TestExecBackendSeedNonZeroExitFails(t *testing.T) {
	b := testExecBackend(t)
	host := "badseed.example.com"
	stageSeedScript(t, b, host, "v1", "echo 'schema is corrupt' >&2\nexit 1\n")

	err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`})
	if err == nil {
		t.Fatal("a seed exiting 1 was reported as success")
	}
	for _, want := range []string{host, "exit status 1", "schema is corrupt"} {
		if !strings.Contains(err.Error(), want) {
			t.Fatalf("seed error = %q, want it to contain %q", err, want)
		}
	}
}

// Unparseable seed argv is an ERROR, never a silent skip: a route whose seed
// column got mangled must fail loudly at the first wake rather than boot an
// empty instance that looks fine.
func TestExecBackendSeedRejectsMalformedArgv(t *testing.T) {
	b := testExecBackend(t)
	host := "malformed.example.com"
	argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")

	err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `seed --fixtures`})
	if err == nil {
		t.Fatal("malformed seed argv was silently skipped")
	}
	if !strings.Contains(err.Error(), host) {
		t.Fatalf("seed parse error = %q, want it to name the host", err)
	}
	if _, statErr := os.Stat(argvPath); statErr == nil {
		t.Fatal("the binary was execed despite unparseable seed argv")
	}
}

// An empty argv array is a no-op, not an exec of the bare binary (defensive:
// doWake only calls Seed for a non-empty route seed).
func TestExecBackendSeedEmptyArgvRunsNothing(t *testing.T) {
	b := testExecBackend(t)
	host := "empty.example.com"
	argvPath, _ := stageRecordingLiveFullEnv(t, b, host, "v1")

	if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `[]`}); err != nil {
		t.Fatalf("empty seed argv: %v", err)
	}
	if _, statErr := os.Stat(argvPath); statErr == nil {
		t.Fatal("an empty seed argv execed the binary")
	}
}

// A seed that never returns is bounded by seedTimeout — a wedged seed must
// fail the wake, not hold the lease until the wake's own 5m budget expires.
func TestExecBackendSeedIsTimeBounded(t *testing.T) {
	b := testExecBackend(t)
	host := "hangs.example.com"
	stageSeedScript(t, b, host, "v1", "sleep 30\n")

	prevTimeout, prevGrace := seedTimeout, seedDrainGrace
	seedTimeout, seedDrainGrace = 200*time.Millisecond, 200*time.Millisecond
	t.Cleanup(func() { seedTimeout, seedDrainGrace = prevTimeout, prevGrace })

	start := time.Now()
	err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`})
	if err == nil {
		t.Fatal("a hanging seed was reported as success")
	}
	if !strings.Contains(err.Error(), "timed out") {
		t.Fatalf("seed error = %q, want it to say the seed timed out", err)
	}
	// The bound that matters: the script sleeps 30s, and a seed still running
	// when Seed returns would be holding the lease behind a wake that already
	// failed.
	if elapsed := time.Since(start); elapsed > 5*time.Second {
		t.Fatalf("seed took %s — the timeout did not bound it", elapsed)
	}
}

// A missing live binary is the same loud failure the spawn path gives: the
// seed goes through ensureLiveLink, so a host with nothing staged never
// silently "seeds".
func TestExecBackendSeedRequiresLiveBinary(t *testing.T) {
	b := testExecBackend(t)
	err := b.Seed(context.Background(), registry.Route{Host: "unstaged.example.com", Kind: "instance", Seed: `["seed"]`})
	if err == nil {
		t.Fatal("seed with no live binary reported success")
	}
	if !strings.Contains(err.Error(), "unstaged.example.com") {
		t.Fatalf("seed error = %q, want it to name the host", err)
	}
}

// A crashed seed takes its half-built database with it. Otherwise the next
// wake stats a file that exists, calls it a warm cache, reports NOT fresh, and
// boots the instance on a half-seeded database — the exact outcome failing the
// wake was meant to prevent.
func TestExecBackendSeedFailureRemovesHalfBuiltDB(t *testing.T) {
	b := testExecBackend(t)
	host := "crashed.example.com"
	// argv is: seed --db <path>  →  $3 is the database path.
	stageSeedScript(t, b, host, "v1", "printf half > \"$3\"\necho 'died mid-schema' >&2\nexit 1\n")

	if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err == nil {
		t.Fatal("a seed exiting 1 was reported as success")
	}
	if _, err := os.Stat(filepath.Join(b.DataDir, host+".db")); err == nil {
		t.Fatal("a failed seed left its half-built database behind — the next wake would treat it as a warm cache")
	}
}

// The cleanup is guarded: a database that already existed when Seed was called
// is never ours to delete (the wake path can't produce this — Restore reported
// fresh — but a direct call must not be able to destroy live data).
func TestExecBackendSeedFailureKeepsPreexistingDB(t *testing.T) {
	b := testExecBackend(t)
	host := "existing-db.example.com"
	stageSeedScript(t, b, host, "v1", "echo nope >&2\nexit 1\n")
	if err := os.WriteFile(filepath.Join(b.DataDir, host+".db"), []byte("live data"), 0o600); err != nil {
		t.Fatal(err)
	}

	if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err == nil {
		t.Fatal("a seed exiting 1 was reported as success")
	}
	got, err := os.ReadFile(filepath.Join(b.DataDir, host+".db"))
	if err != nil || string(got) != "live data" {
		t.Fatalf("pre-existing database = %q, %v; want it untouched by a failed seed", got, err)
	}
}

// A seed that exits 0 but backgrounds a child holding its stderr pipe open is
// a SUCCESS. os/exec's WaitDelay fires on the success path too — Run returns
// ErrWaitDelay instead of nil — and treating that as a failure would fail the
// wake and delete the database the seed had just correctly built, leaving the
// host fresh again on the next wake: permanently un-wakeable.
func TestExecBackendSeedSucceedsDespiteLingeringChild(t *testing.T) {
	b := testExecBackend(t)
	host := "leaky-seed.example.com"
	// argv is: seed --db <path> → $3 is the database path. The backgrounded
	// sleeper inherits stderr and holds the pipe open past the drain grace.
	stageSeedScript(t, b, host, "v1", "printf seeded > \"$3\"\nsleep 5 &\nexit 0\n")

	prevTimeout, prevGrace := seedTimeout, seedDrainGrace
	seedTimeout, seedDrainGrace = 10*time.Second, 200*time.Millisecond
	t.Cleanup(func() { seedTimeout, seedDrainGrace = prevTimeout, prevGrace })

	if err := b.Seed(context.Background(), registry.Route{Host: host, Kind: "instance", Seed: `["seed"]`}); err != nil {
		t.Fatalf("Seed = %v, want nil — the seed exited 0; a leaked output pipe is not a failed seed", err)
	}
	got, err := os.ReadFile(filepath.Join(b.DataDir, host+".db"))
	if err != nil || string(got) != "seeded" {
		t.Fatalf("seeded database = %q, %v; want it kept — deleting it would make the host fresh again next wake", got, err)
	}
}

// The delete guard is fail-safe by construction: it arms ONLY on a definite
// "not there". Asserted on the rule itself, because the interesting inputs —
// a stat that failed for a reason other than absence — are exactly the ones an
// integration test can't stage reliably.
func TestSeedCreatedTheDB(t *testing.T) {
	for _, tc := range []struct {
		name    string
		statErr error
		want    bool
	}{
		{"database is there", nil, false},
		{"definitely absent", &os.PathError{Op: "stat", Path: "/data/x.db", Err: syscall.ENOENT}, true},
		{"bare ErrNotExist", os.ErrNotExist, true},
		{"unreadable", &os.PathError{Op: "stat", Path: "/data/x.db", Err: syscall.EACCES}, false},
		{"path not traversable", &os.PathError{Op: "stat", Path: "/data/x.db", Err: syscall.ENOTDIR}, false},
		{"i/o error", &os.PathError{Op: "stat", Path: "/data/x.db", Err: syscall.EIO}, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if got := seedCreatedTheDB(tc.statErr); got != tc.want {
				t.Fatalf("seedCreatedTheDB(%v) = %v, want %v — an inconclusive stat must never arm the removal of a database", tc.statErr, got, tc.want)
			}
		})
	}
}

// --- Instance-socket mode normalization (issue #103) -------------------------

// TestMain lets this test binary double as a fake instance app: Start execs
// the live/<host> symlink, and stageServingLive points that symlink at the
// test binary itself with ACTIVATOR_FAKE_APP=1 in the inherited environment
// (childEnv strips only CARLOS_* vars, so it flows through to the child).
// The shell-script recorders elsewhere in this file can't answer the health
// probe — serving HTTP on a unix socket needs a real program, and the test
// binary is the one program every `go test` run is guaranteed to have.
func TestMain(m *testing.M) {
	// This binary places processes on a real cgroupfs in its kernel tests
	// (TestGenerationStartRefusedBesideADirectParentMember,
	// TestClaimContainmentCleanupRespawn), so it serves the placement shim
	// like the carlos binary does (cgroup.RunShimIfRequested).
	cgroup.RunShimIfRequested()
	if os.Getenv("ACTIVATOR_FAKE_APP") == "1" {
		fakeAppMain()
		return
	}
	// A worker's wake waits out its settle window unless the child reports
	// ready, and no fake child does. Shrink it once for the whole package —
	// the tests that are ABOUT the window set their own value and restore it
	// (spawnHealthTimeout's pattern in backend_unit_test.go).
	sidecarSettle = 20 * time.Millisecond
	os.Exit(m.Run())
}

// Which tests call t.Parallel(). A parallel test runs beside the other
// parallel tests once every sequential one has finished, so it may not
// write package state (a shrunk timeout, a hook, the generations switch,
// the log's output), set the environment or the working directory, depend
// on time it measures itself (time.Since), or use the /proc child waits in
// childwait_test.go, which count every child of this binary. A test that
// does any of these stays sequential. The tests marked parallel were
// checked for all of it, through every test helper they call; a test that
// later starts shrinking a package var or setting the environment must
// drop its t.Parallel().

// fakeAppMain is the app half: bind the --socket, tighten it to 0600 — the
// exact titogo habit that stranded the sibling router in issue #103 — and
// serve 200s until killed (500s under ACTIVATOR_FAKE_APP_UNHEALTHY, the
// never-becomes-healthy app).
func fakeAppMain() {
	var socket string
	for i, a := range os.Args {
		if a == "--socket" && i+1 < len(os.Args) {
			socket = os.Args[i+1]
		}
	}
	if socket == "" {
		os.Exit(2)
	}
	ln, err := net.Listen("unix", socket)
	if err != nil {
		os.Exit(1)
	}
	if err := os.Chmod(socket, 0o600); err != nil {
		os.Exit(1)
	}
	status := http.StatusOK
	if os.Getenv("ACTIVATOR_FAKE_APP_UNHEALTHY") == "1" {
		status = http.StatusInternalServerError
	}
	_ = http.Serve(ln, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(status)
	}))
}

// stageServingLive points live/<host> at the test binary itself (see
// TestMain) so Start spawns a child that actually binds and serves.
func stageServingLive(t *testing.T, b *ExecBackend, host string) {
	t.Helper()
	self, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	t.Setenv("ACTIVATOR_FAKE_APP", "1")
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	os.Remove(b.liveLink(host))
	if err := os.Symlink(self, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}
}

// TestExecBackendStartNormalizesSocketMode pins the issue #103 contract: the
// mode an instance socket carries after a successful wake is the PLATFORM's
// (0660 — group members, i.e. enrolled carlos-sockets consumers, may dial),
// never the app's own umask or chmod habit. The fake app deliberately
// tightens its socket to 0600 after binding; Start must win.
func TestExecBackendStartNormalizesSocketMode(t *testing.T) {
	requireUnixSockets(t)
	b := testExecBackend(t)
	host := "mode.example.com"
	stageServingLive(t, b, host)

	rt := registry.Route{Host: host, Kind: "instance", Socket: tmpSock(t)}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	proc, _, err := b.Start(ctx, rt)
	if err != nil {
		t.Fatalf("Start: %v", err)
	}
	defer proc.Stop(context.Background())

	fi, err := os.Stat(rt.Socket)
	if err != nil {
		t.Fatal(err)
	}
	if fi.Mode().Perm() != 0o660 {
		t.Fatalf("socket mode after healthy Start = %o, want 660 — an app's own chmod must not strand a sibling consumer", fi.Mode().Perm())
	}
}

// TestExecBackendFailedWakeDoesNotNormalizeSocketMode pins the other half of
// the contract: normalization happens only AFTER a successful health wait —
// a wake that fails must leave the socket exactly as the app left it. The
// fake app binds and tightens to 0600 but never answers healthily.
func TestExecBackendFailedWakeDoesNotNormalizeSocketMode(t *testing.T) {
	requireUnixSockets(t)
	b := testExecBackend(t)
	host := "unhealthy.example.com"
	stageServingLive(t, b, host)
	t.Setenv("ACTIVATOR_FAKE_APP_UNHEALTHY", "1")

	old := spawnHealthTimeout
	spawnHealthTimeout = 500 * time.Millisecond
	defer func() { spawnHealthTimeout = old }()

	rt := registry.Route{Host: host, Kind: "instance", Socket: tmpSock(t)}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if proc, _, err := b.Start(ctx, rt); err == nil {
		defer proc.Stop(context.Background())
		t.Fatal("Start succeeded for a never-healthy app")
	}
	fi, err := os.Stat(rt.Socket)
	if err != nil {
		t.Fatal(err)
	}
	if fi.Mode().Perm() != 0o600 {
		t.Fatalf("socket mode after failed wake = %o, want the app's own 600 — a failed wake must not widen anything", fi.Mode().Perm())
	}
}

// TestExecBackendDoesNotNormalizeSquatterSocket pins the only-our-child
// guard: a stray supervisor squatting the route's socket (the #17 advisory's
// subject) can answer the health probe while the spawned child — which lost
// the bind race and exited — cannot. Start must not widen the squatter's
// socket. The squatter here holds its 200 until the child has verifiably
// exited, making the race the guard closes deterministic.
func TestExecBackendDoesNotNormalizeSquatterSocket(t *testing.T) {
	t.Parallel()
	requireUnixSockets(t)
	b := testExecBackend(t)
	host := "squatted.example.com"
	marker := filepath.Join(t.TempDir(), "child-ran")
	// The child: proof-of-life marker, then immediate exit — the shape of a
	// child that could not bind.
	script := "#!/bin/sh\ntouch '" + marker + "'\nexit 0\n"
	vdir := filepath.Join(b.OptDir, "versions", "v1")
	if err := os.MkdirAll(vdir, 0o755); err != nil {
		t.Fatal(err)
	}
	bin := filepath.Join(vdir, "app")
	if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(bin, b.liveLink(host)); err != nil {
		t.Fatal(err)
	}

	sock := tmpSock(t)
	ln, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatal(err)
	}
	defer ln.Close()
	if err := os.Chmod(sock, 0o600); err != nil {
		t.Fatal(err)
	}
	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		// Answer only once the child is provably dead: marker written, plus
		// a grace for the exit itself to be reaped.
		deadline := time.Now().Add(5 * time.Second)
		for time.Now().Before(deadline) {
			if _, err := os.Stat(marker); err == nil {
				break
			}
			time.Sleep(10 * time.Millisecond)
		}
		time.Sleep(300 * time.Millisecond)
		w.WriteHeader(http.StatusOK)
	})}
	go srv.Serve(ln)
	defer srv.Close()

	rt := registry.Route{Host: host, Kind: "instance", Socket: sock}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	proc, _, err := b.Start(ctx, rt)
	if err != nil {
		t.Fatalf("Start: %v (the squatter answers healthily; the wake itself succeeds)", err)
	}
	defer proc.Stop(context.Background())

	fi, err := os.Stat(rt.Socket)
	if err != nil {
		t.Fatal(err)
	}
	if fi.Mode().Perm() != 0o600 {
		t.Fatalf("squatter socket mode after Start = %o, want its own 600 — a socket our child did not bind is never ours to widen", fi.Mode().Perm())
	}
}

// The control socket must OUTLIVE the drain. Stop used to close it before the
// child was even signalled, which made the spec's §7.1 story — an app that
// re-asserts its once-timers as it winds down, over the §7.2 verbs on
// CARLOS_CONTROL_SOCKET — a connection-refused during exactly the window the
// app was told to use it.
//
// The test drives the real childProc: a child that traps SIGTERM and stays up
// until the test releases it, a real edge.ControlListener with a verb
// mounted, and a request issued strictly between "SIGTERM delivered" and
// "child exited". It also pins the other half — that the listener is still
// closed, on this path, once the child is gone.
func TestChildProcStopKeepsTheControlSocketUpUntilTheChildExits(t *testing.T) {
	t.Parallel()
	requireUnixSockets(t)
	dir := t.TempDir()
	ready := filepath.Join(dir, "ready")     // the child's SIGTERM trap is installed
	termed := filepath.Join(dir, "termed")   // the child saw SIGTERM
	release := filepath.Join(dir, "release") // the test lets the child finish
	script := filepath.Join(dir, "app")
	// The ready marker is touched AFTER the trap is installed, and the test
	// waits for it before signalling: a SIGTERM that arrives in the shell's
	// first few instructions takes the default action and kills the child
	// outright, and there is no drain window to observe.
	body := "#!/bin/sh\n" +
		"trap \"touch '" + termed + "'\" TERM\n" +
		"touch '" + ready + "'\n" +
		"while [ ! -f '" + release + "' ]; do sleep 0.02; done\n" +
		"exit 0\n"
	if err := os.WriteFile(script, []byte(body), 0o755); err != nil {
		t.Fatal(err)
	}

	ctlSock := tmpSock(t)
	ctl, err := edge.ListenControl(ctlSock, "drain.example.com", nil, func(mux *http.ServeMux) {
		mux.HandleFunc("GET /schedule", func(w http.ResponseWriter, r *http.Request) {
			w.WriteHeader(http.StatusOK)
		})
	})
	if err != nil {
		t.Fatalf("ListenControl: %v", err)
	}

	cmd := exec.Command(script)
	if err := cmd.Start(); err != nil {
		ctl.Close()
		t.Fatalf("start the fake child: %v", err)
	}
	// The child traps TERM and polls a file inside t.TempDir(). Any t.Fatalf
	// before the release file is written — the failure this test exists to
	// produce, among others — leaves it spinning on a directory the cleanup
	// then deletes, immune to the very signal this test is about: only
	// SIGKILL ends it, and a red CI run would leak one per attempt.
	t.Cleanup(func() { _ = cmd.Process.Kill() })
	p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl}
	go func() {
		p.err = cmd.Wait()
		close(p.done)
	}()

	waitForFile(t, ready, 10*time.Second)

	stopped := make(chan error, 1)
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		stopped <- p.Stop(ctx)
	}()

	// Strictly inside the drain: SIGTERM has been delivered (the child's own
	// trap says so) and the child has not exited.
	waitForFile(t, termed, 10*time.Second)
	if p.Exited() {
		t.Fatal("the child exited before the drain window could be observed — the test proves nothing")
	}
	req, err := http.NewRequest("GET", "http://carlos/schedule", nil)
	if err != nil {
		t.Fatal(err)
	}
	resp, err := socketClient(ctlSock).Do(req)
	if err != nil {
		t.Fatalf("control request during the drain: %v — the socket must still answer while the child drains", err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("control request during the drain = %d, want 200", resp.StatusCode)
	}

	if err := os.WriteFile(release, nil, 0o600); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-stopped:
		if err != nil {
			t.Fatalf("Stop: %v", err)
		}
	case <-time.After(30 * time.Second):
		t.Fatal("Stop never returned after the child was released")
	}
	// The close is only deferred, not dropped: a listener left bound to a
	// socket with no process behind it is the #129 stale-socket wake trap.
	if _, err := os.Stat(ctlSock); !os.IsNotExist(err) {
		t.Fatalf("control socket after Stop returned: stat err = %v, want it unlinked", err)
	}
}

// The already-exited path takes an early return, and the listener close has
// to survive it: an instance whose child crashed on its own still leaves the
// agent holding a bound socket nothing is behind.
func TestChildProcStopClosesTheControlSocketWhenTheChildIsAlreadyGone(t *testing.T) {
	requireUnixSockets(t)
	ctlSock := tmpSock(t)
	ctl, err := edge.ListenControl(ctlSock, "gone.example.com", nil, nil)
	if err != nil {
		t.Fatalf("ListenControl: %v", err)
	}
	cmd := exec.Command("/bin/sh", "-c", "exit 0")
	if err := cmd.Start(); err != nil {
		ctl.Close()
		t.Fatal(err)
	}
	p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl}
	p.err = cmd.Wait()
	close(p.done) // the child is gone before Stop is ever called

	if err := p.Stop(context.Background()); err != nil {
		t.Fatalf("Stop on an already-exited child: %v", err)
	}
	if _, err := os.Stat(ctlSock); !os.IsNotExist(err) {
		t.Fatalf("control socket after Stop on an exited child: stat err = %v, want it unlinked", err)
	}
}

func TestChildEnvKeepsOneEntryPerKeyLastWins(t *testing.T) {
	t.Setenv("PATH", "/usr/bin")
	env := childEnv(map[string]string{"PATH": "/tenant/bin", "A": "1"})
	seen := map[string]string{}
	for _, kv := range env {
		k, v, _ := strings.Cut(kv, "=")
		if _, dup := seen[k]; dup {
			t.Fatalf("duplicate key %s in %v", k, env)
		}
		seen[k] = v
	}
	if seen["PATH"] != "/tenant/bin" {
		t.Errorf("PATH = %q, want the tenant's value to win over the inherited one", seen["PATH"])
	}
	if seen["A"] != "1" {
		t.Errorf("A = %q", seen["A"])
	}
}

// A Proc whose Stop reports ErrStillRunning must not have its lease
// released: stop() keeps the entry in stopping and the sweeper finishes it
// once the proc reports exited (design §6.3, codex R2-5).
func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	b.procStuck = true // Start returns a fakeProc whose Stop reports ErrStillRunning until exit()
	rt := testRoute("h.example.com")
	arrangeAwake(t, a, rt) // wake rt and release the request, as the idle-hibernate test does

	clk.advance(2 * time.Minute)
	a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning
	p := b.lastProc()
	waitFor(t, "the stuck stop to be attempted", func() bool { return p.stopCalls() >= 1 })
	if n := b.releaseCount(); n != 0 {
		t.Fatalf("lease released %d times while the stop is unproven", n)
	}
	if a.entryCount() != 1 {
		t.Fatal("entry forgotten before the stop was proven")
	}
	p.exit()
	// Drive the sweep from inside the wait itself: finishing a pendingStop
	// entry now retries proc.Stop (design §6.3) on the sweeper's own
	// cadence, so a single sweepOnce() right after exit() is a race against
	// exactly when the sweeper would next notice. Polling sweepOnce()
	// alongside the release count removes that race instead of hoping one
	// call lands after exit() is visible.
	waitFor(t, "the lease to release once termination is proven", func() bool {
		a.sweepOnce()
		return b.releaseCount() == 1
	})
}

// finishPendingStops must retry the stop itself while the process is still
// running, not only once it has already exited (design §6.3: "sweepOnce
// retries proc.Stop on its cadence until done closes"). Without the retry, a
// stop whose first Control.Stop call itself failed would never ask again.
func TestFinishPendingStopsRetriesTheStopWhileStillRunning(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	b.procStuck = true // Stop reports ErrStillRunning until exit(), and this proc never exits here
	rt := testRoute("stuck.example.com")
	arrangeAwake(t, a, rt)

	clk.advance(2 * time.Minute)
	a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning, entry goes pendingStop
	p := b.lastProc()
	waitFor(t, "the first stop attempt", func() bool { return p.stopCalls() >= 1 })

	// Immediately after: no retry yet, the entry's own actStopTimeout budget
	// has not passed since the first attempt.
	a.sweepOnce()
	if n := p.stopCalls(); n != 1 {
		t.Fatalf("stop retried before actStopTimeout elapsed (stopCalls=%d)", n)
	}

	clk.advance(actStopTimeout + time.Second)
	waitFor(t, "a second stop attempt once actStopTimeout has passed", func() bool {
		a.sweepOnce()
		return p.stopCalls() >= 2
	})
	if b.releaseCount() != 0 {
		t.Fatal("lease released while the process still reports running")
	}
	if a.entryCount() != 1 {
		t.Fatal("entry forgotten while the process still reports running")
	}
}

// The deferred Shutdown drain (ledger T5): a pendingStop entry whose process
// finally exits DURING Shutdown still gets its final checkpoint and lease
// release before Shutdown returns, via the same finishPendingStops pass
// Shutdown's own drain loop re-runs on its poll ticker.
func TestShutdownDrainsAPendingStopThatExitsMidShutdown(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	b.procStuck = true
	rt := testRoute("shutdown-pending.example.com")
	arrangeAwake(t, a, rt)

	clk.advance(2 * time.Minute)
	a.sweepOnce() // stop() runs, Stop reports ErrStillRunning, entry goes pendingStop
	p := b.lastProc()
	waitFor(t, "the stop to be attempted", func() bool { return p.stopCalls() >= 1 })

	// The process exits on its own shortly after Shutdown begins draining.
	// Shutdown's 100ms poll ticker is what has to notice it, not this test.
	go func() {
		time.Sleep(150 * time.Millisecond)
		p.exit()
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.Shutdown(ctx)

	if got := b.releaseCount(); got != 1 {
		t.Fatalf("lease releases after Shutdown = %d, want exactly 1", got)
	}
	if a.entryCount() != 0 {
		t.Fatal("Shutdown returned with a pending-stop entry still on the books")
	}
}

// The sidecar restart must not hold the activator mutex across the start
// call (design §6.7, codex R2-6): a stalled sidecar start does not block
// another host's wake.
func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
	a, b, clk, _ := testActivator(t, time.Hour)
	b.sidecarStaged = true

	// wake a.example.com first, ungated. The gate below is for the
	// sweeper's RESTART call, not this initial co-spawn.
	arrangeAwake(t, a, testRoute("a.example.com"))
	b.lastSidecar().exit() // the sweeper now wants a restart

	gate := make(chan struct{})
	entered := make(chan struct{})
	b.mu.Lock()
	b.sidecarGate = gate // StartSidecar blocks on this channel when non-nil
	b.sidecarStartEntered = entered
	b.mu.Unlock()
	clk.advance(time.Minute)
	a.sweepOnce() // launches the restart; it blocks inside StartSidecar on gate
	<-entered     // wait for the restart's call to actually begin, not a guessed sleep
	// host b has no sidecar of its own. HasSidecar false keeps its wake's
	// own co-spawn (unrelated to what's under test) off the same gate.
	b.mu.Lock()
	b.sidecarStaged = false
	b.mu.Unlock()
	done := make(chan struct{})
	go func() {
		// wake b.example.com
		release, err := a.beginRequest(context.Background(), testRoute("b.example.com"))
		if err != nil {
			t.Errorf("beginRequest b.example.com: %v", err)
			return
		}
		release()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("a second host's wake blocked behind a stalled sidecar start")
	}
	close(gate)
}

// doWake runs on the committed row, not the copy the edge supplied
// (design §6.8, codex R8-1).
func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
	a, b, _, reg := testActivator(t, time.Hour)
	a.Routes = reg
	old := testRoute("h.example.com")
	fresh := old
	fresh.DBPath = "/data/instances/h.example.com/h.example.com.db"
	fresh.Socket = "/run/carlos/instances/h.example.com/app.sock"
	// insert fresh into reg the way the file's registry-backed tests do
	if err := reg.Upsert(fresh); err != nil {
		t.Fatalf("Upsert: %v", err)
	}
	arrangeAwake(t, a, old) // wake with old (the edge's stale copy)
	if got := b.lastStartedRoute().DBPath; got != fresh.DBPath {
		t.Fatalf("Start saw DBPath %q, want the registry's %q", got, fresh.DBPath)
	}
	if got := b.lastStartedRoute().Socket; got != fresh.Socket {
		t.Fatalf("Start saw Socket %q, want the registry's %q", got, fresh.Socket)
	}
}

// A restart whose StartSidecar call is still stalled when its host stops
// (Task 8, F6 plan 5, flag off): stop() waits for it only so long
// (sidecarStartJoinWait), and then the stop stays pending, with the lease,
// until the start returns. The sidecar the start brought back is stopped,
// and only then is the host cgroup proven empty and the lease released. A
// stale restart therefore can no longer land on a fresh wake of the host:
// that wake cannot begin until this entry is gone. The restart still
// compares entry identity (the *actEntry captured at launch) rather than a
// counter, and the fresh wake below still gets its own sidecar.
func TestAStopKeepsTheLeaseUntilAnInFlightSidecarStartReturns(t *testing.T) {
	prev := sidecarStartJoinWait
	sidecarStartJoinWait = 50 * time.Millisecond
	t.Cleanup(func() { sidecarStartJoinWait = prev })
	a, b, _, _ := testActivator(t, time.Hour)
	h := &hostEmptyFake{fakeBackend: b}
	a.backend = h
	b.sidecarStaged = true
	host := "stale.example.com"

	arrangeAwake(t, a, testRoute(host))
	h.reset() // the wake's own proof (finding 4) is not the stop's
	first := b.lastSidecar()
	first.exit() // the sweeper now wants a restart

	gate := make(chan struct{})
	entered := make(chan struct{})
	b.mu.Lock()
	b.sidecarGate = gate
	b.sidecarStartEntered = entered
	b.mu.Unlock()
	a.sweepOnce() // launches the restart; it blocks inside StartSidecar on gate
	<-entered

	rep, inst := b.lastRep(), b.lastProc()
	rep.setOnClose(func() {
		stale := b.lastSidecar()
		if stale == first || stale.stopCalls() < 1 {
			t.Error("the lease was released before the in-flight start's sidecar was stopped")
		}
		ev := b.events()
		stopped := slices.Index(ev, "stop:sidecar:late")
		if stopped < 0 || !slices.Contains(ev[stopped:], "emptyhost") {
			t.Errorf("events %v: want the late sidecar stopped, then the host proven empty, before the release", ev)
		}
	})
	if !a.Bounce(host) {
		t.Fatal("Bounce did not find the awake entry")
	}
	waitFor(t, "the instance to stop", inst.isStopped)
	time.Sleep(4 * sidecarStartJoinWait) // well past stop()'s own wait
	for range 3 {
		a.sweepOnce()
	}
	if rep.isClosed() || a.entryCount() != 1 {
		t.Fatalf("closed=%v entries=%d while a sidecar start was in flight: the lease went first", rep.isClosed(), a.entryCount())
	}
	if n := h.count(); n != 0 {
		t.Fatalf("the host cgroup was proven %d times before the in-flight start returned", n)
	}

	b.mu.Lock()
	b.sidecarGate = nil // the fresh wake's own co-spawn must not block on it too
	b.mu.Unlock()
	b.lateSidecarLabel("sidecar:late")
	close(gate) // let the stale StartSidecar call finally return
	waitFor(t, "the stop to finish once the start returned", func() bool {
		a.sweepOnce()
		return rep.isClosed() && a.entryCount() == 0
	})
	b.lateSidecarLabel("")

	// A fresh wake of the same host gets its own sidecar.
	arrangeAwake(t, a, testRoute(host))
	freshSidecar := b.lastSidecar()
	a.mu.Lock()
	installed := a.entries[host].sidecar
	a.mu.Unlock()
	if installed != freshSidecar || freshSidecar.label != "sidecar" {
		t.Fatal("the fresh wake did not get its own sidecar")
	}
}

// seedCountingBackend is fakeBackend plus the optional seed recorder.
type seedCountingBackend struct {
	fakeBackend
	placed     []string
	rep        Rep
	replicated int // Replicate calls seen when SeedPlaced ran
	err        error
}

func (b *seedCountingBackend) SeedPlaced(_ context.Context, rt registry.Route, rep Rep) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.placed = append(b.placed, rt.Host)
	b.rep, b.replicated = rep, len(b.syncIntervals)
	return b.err
}

func TestDoWakeRecordsSeedPlacementAfterReplication(t *testing.T) {
	t.Parallel()
	for _, recErr := range []error{nil, errors.New("record unreachable")} {
		b := &seedCountingBackend{err: recErr}
		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)
		rt := testRoute("shop.example.com")
		rt.SeedSource = seeds.RowSeed{ID: "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa", Kind: seeds.KindReplica, State: seeds.StateReady}.Encode()
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
		// A failed record never fails the wake: the instance serves.
		arrangeAwake(t, a, rt)
		if len(b.placed) != 1 || b.placed[0] != rt.Host {
			t.Fatalf("SeedPlaced calls = %q, want one, after replication started", b.placed)
		}
		if len(b.reps) != 1 || b.rep != Rep(b.reps[0]) || b.replicated != 1 {
			t.Fatalf("SeedPlaced saw rep %v after %d replications; want the wake's own, after Replicate", b.rep, b.replicated)
		}
		if !a.Owns(rt.Host) {
			t.Fatalf("recorder error %v: the wake did not serve", recErr)
		}
	}
}

// hostEmptyFake is fakeBackend with Task 8's host-cgroup proof
// (hostEmptier), the flag-off terminal stop's: err is its answer, and each
// call is counted and logged as "emptyhost" in the backend's order log.
type hostEmptyFake struct {
	*fakeBackend
	mu     sync.Mutex
	err    error
	calls  int
	passes int // calls that answered nil
}

func (h *hostEmptyFake) EmptyHost(context.Context, registry.Route) error {
	h.mu.Lock()
	h.calls++
	err := h.err
	if err == nil {
		h.passes++
	}
	h.mu.Unlock()
	h.record("emptyhost")
	return err
}

func (h *hostEmptyFake) passed() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.passes
}

func (h *hostEmptyFake) set(err error) {
	h.mu.Lock()
	h.err = err
	h.mu.Unlock()
}

func (h *hostEmptyFake) count() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.calls
}

// reset forgets the proofs made so far: a stop test calls it after its wake,
// whose own flag-off proof (finding 4) it does not count.
func (h *hostEmptyFake) reset() {
	h.mu.Lock()
	h.calls, h.passes = 0, 0
	h.mu.Unlock()
}

// Task 8 fix round 1, finding 4: after an agent restart the in-memory
// pending stop and kill marks are gone, so a flag-off wake proves the host
// cgroup empty itself, under its lease and before Restore. A proof that
// fails refuses the wake before anything is restored or started, and keeps
// the lease (a leftover may still write the database). One that holds lets
// the wake go on.
func TestAFlagOffWakeProvesTheHostCgroupBeforeItRestores(t *testing.T) {
	a, b, _, _ := testActivator(t, time.Minute)
	h := &hostEmptyFake{fakeBackend: b}
	a.backend = h
	h.set(errors.New("a leftover still populates the host cgroup"))
	rt := testRoute("leftover.example.com")

	if _, err := a.beginRequest(context.Background(), rt); !errors.Is(err, errHostUnproven) {
		t.Fatalf("wake = %v, want errHostUnproven", err)
	}
	if _, restores, starts := b.counts(); restores != 0 || starts != 0 {
		t.Fatalf("restores=%d starts=%d over a host cgroup not proven empty, want neither", restores, starts)
	}
	if got := b.releaseCount(); got != 0 {
		t.Fatalf("lease releases = %d, want 0: a leftover may still write the database", got)
	}
	if got := memberSafeWakeReason(fmt.Errorf("leftover.example.com: refusing to wake: %w", errHostUnproven)); got != memberSafeWakeReason(ErrFenced) {
		t.Fatalf("member reads %q, want the fence's existing text", got)
	}

	h.set(nil)
	if _, err := a.beginRequest(context.Background(), rt); err != nil {
		t.Fatalf("wake = %v once the proof holds", err)
	}
	ev := b.events()
	last := func(name string) int {
		for i := len(ev) - 1; i >= 0; i-- {
			if ev[i] == name {
				return i
			}
		}
		return -1
	}
	proven, restored := last("emptyhost"), last("restore")
	if proven < 0 || restored < proven {
		t.Fatalf("events %v: want the host proven empty before the restore", ev)
	}
}

// No delegated cgroup: the wake goes on as today. A row whose child is not
// in carlos-exec@<host> (nested, unit-backed) asks no proof, and neither
// does a wake with generations on, whose proofs are its own.
func TestAWakeAsksTheHostProofOnlyFlagOffForAnExecChild(t *testing.T) {
	t.Parallel()
	t.Run("no delegation", func(t *testing.T) {
		a, b, _, _ := testActivator(t, time.Minute)
		h := &hostEmptyFake{fakeBackend: b}
		h.set(cgroup.ErrNoDelegation)
		a.backend = h
		if _, err := a.beginRequest(context.Background(), testRoute("undelegated-wake.example.com")); err != nil {
			t.Fatalf("wake = %v, want today's wake with no delegation", err)
		}
		if h.count() != 1 {
			t.Fatalf("proofs = %d, want 1", h.count())
		}
	})
	for _, tc := range []struct {
		name  string
		setup func(*Service, *fakeBackend, *registry.Route)
	}{
		{"nested", func(_ *Service, b *fakeBackend, _ *registry.Route) {
			b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
		}},
		{"unit-backed", func(_ *Service, _ *fakeBackend, rt *registry.Route) { rt.Backing = registry.BackingUnit }},
		{"generations on", func(a *Service, _ *fakeBackend, _ *registry.Route) { a.gens = true }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			a, b, _, _ := testActivator(t, time.Minute)
			h := &hostEmptyFake{fakeBackend: b}
			h.set(errors.New("must not be asked"))
			a.backend = h
			rt := testRoute("no-host-proof.example.com")
			tc.setup(a, b, &rt)
			_, _ = a.beginRequest(context.Background(), rt)
			if n := h.count(); n != 0 {
				t.Fatalf("host proofs = %d, want none", n)
			}
		})
	}
}

// lateSidecarLabel labels every sidecar StartSidecar returns from now on.
func (b *fakeBackend) lateSidecarLabel(label string) {
	b.mu.Lock()
	b.sidecarLabel = label
	b.mu.Unlock()
}

// Task 8 (F6, flag off): an idle stop proves the host cgroup empty before
// the lease goes. While the proof fails the stop stays pending with the
// lease held; the sweep retries, and the lease goes only after a proof
// that holds.
func TestAFlagOffStopKeepsTheLeaseUntilTheHostCgroupIsProvenEmpty(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	h := &hostEmptyFake{fakeBackend: b}
	a.backend = h
	arrangeAwake(t, a, testRoute("f6.example.com"))
	h.reset() // the wake's own proof (finding 4) is not the stop's
	h.set(errors.New("the host cgroup still holds a process"))

	clk.advance(2 * time.Minute)
	a.sweepOnce()
	waitFor(t, "the stop to go pending on the proof", func() bool {
		a.mu.Lock()
		defer a.mu.Unlock()
		e := a.entries["f6.example.com"]
		return h.count() >= 1 && e != nil && e.pendingStop && !e.genReaping
	})
	rep := b.lastRep()
	if rep.isClosed() || a.entryCount() != 1 {
		t.Fatal("the lease was released over a host cgroup not proven empty")
	}

	rep.setOnClose(func() {
		if n := h.passed(); n != 1 {
			t.Errorf("passing proofs before the release = %d, want 1", n)
		}
	})
	h.set(nil)
	waitFor(t, "the lease to go once the proof holds", func() bool {
		a.sweepOnce()
		return rep.isClosed() && a.entryCount() == 0
	})
}

// A passing proof releases the lease in the stop itself, after the proof.
func TestAFlagOffStopProvesTheHostCgroupBeforeTheLeaseGoes(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	h := &hostEmptyFake{fakeBackend: b}
	a.backend = h
	arrangeAwake(t, a, testRoute("proven.example.com"))
	h.reset() // the wake's own proof (finding 4) is not the stop's
	rep := b.lastRep()
	var atClose atomic.Int32
	atClose.Store(-1)
	rep.setOnClose(func() { atClose.Store(int32(h.count())) })

	clk.advance(2 * time.Minute)
	a.sweepOnce()
	waitFor(t, "the stop", func() bool { return a.entryCount() == 0 })
	if got := atClose.Load(); got != 1 {
		t.Fatalf("proofs before the release = %d, want 1", got)
	}
}

// No delegated cgroup (cgroup.ErrNoDelegation): the leader's exit is the
// only proof there is (ledger 127), and the lease goes at once.
func TestAFlagOffStopWithNoDelegationReleasesAtOnce(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	h := &hostEmptyFake{fakeBackend: b}
	h.set(cgroup.ErrNoDelegation)
	a.backend = h
	arrangeAwake(t, a, testRoute("undelegated.example.com"))
	h.reset() // the wake's own proof (finding 4) is not the stop's

	clk.advance(2 * time.Minute)
	a.sweepOnce()
	waitFor(t, "the stop, with no sweep behind it", func() bool { return a.entryCount() == 0 })
	if !b.lastRep().isClosed() || h.count() != 1 {
		t.Fatalf("closed=%v proofs=%d, want the lease released after one proof", b.lastRep().isClosed(), h.count())
	}
}

// A nested row's child ran as a unit in its host slice, not in
// carlos-exec@<host>: its stop asks no host proof, decided from the layout
// the wake launched on (actEntry.launchLayout, set at install).
func TestAFlagOffStopOfANestedRowAsksNoHostProof(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
	h := &hostEmptyFake{fakeBackend: b}
	h.set(errors.New("must not be asked"))
	a.backend = h
	arrangeAwake(t, a, testRoute("nested-stop.example.com"))

	clk.advance(2 * time.Minute)
	a.sweepOnce()
	waitFor(t, "the stop", func() bool { return a.entryCount() == 0 })
	if n := h.count(); n != 0 || !b.lastRep().isClosed() {
		t.Fatalf("proofs=%d closed=%v, want no host proof and the lease released", n, b.lastRep().isClosed())
	}
}

// EmptyHost's edges: no delegation is ErrNoDelegation (the caller's
// leader-only proof), and a unit-backed row has no host cgroup of ours.
func TestEmptyHostDispatch(t *testing.T) {
	eb := testExecBackend(t)
	rt := registry.Route{Host: "edge.example.com"}
	if err := eb.EmptyHost(context.Background(), rt); !errors.Is(err, cgroup.ErrNoDelegation) {
		t.Fatalf("EmptyHost with no cgroup manager = %v, want ErrNoDelegation", err)
	}
	sb := &SplitBackend{Exec: eb, Unit: &UnitBackend{}}
	if err := sb.EmptyHost(context.Background(), rt); !errors.Is(err, cgroup.ErrNoDelegation) {
		t.Fatalf("SplitBackend.EmptyHost (exec row) = %v, want the exec side's answer", err)
	}
	rt.Backing = registry.BackingUnit
	if err := sb.EmptyHost(context.Background(), rt); err != nil {
		t.Fatalf("SplitBackend.EmptyHost (unit row) = %v, want nil", err)
	}
}

// A kernel before 5.14 has no cgroup.kill, so the host cgroup cannot be
// killed there at all. EmptyHost answers that as no delegation (the
// leader-only proof, ledger 127), not as an unproven kill that would hold
// every flag-off stop's lease on such a box.
func TestEmptyHostWithNoCgroupKillIsTheLeaderOnlyProof(t *testing.T) {
	eb := testExecBackend(t)
	opened := 0
	eb.openHostCleanup = func(string) (*cgroup.Cleanup, error) {
		opened++
		return nil, fmt.Errorf("%w: open cgroup.kill: no such file or directory", cgroup.ErrNoKill)
	}
	err := eb.EmptyHost(context.Background(), registry.Route{Host: "old-kernel.example.com"})
	if opened != 1 {
		t.Fatalf("host cgroup opened %d times, want 1", opened)
	}
	if !errors.Is(err, cgroup.ErrNoDelegation) || !errors.Is(err, cgroup.ErrNoKill) {
		t.Fatalf("EmptyHost with no cgroup.kill = %v, want ErrNoDelegation naming ErrNoKill", err)
	}
}

// The pending-stop path (a leader that was still running when stop() asked)
// is held to the same proof once the leader exits.
func TestAPendingFlagOffStopProvesTheHostCgroupBeforeTheLeaseGoes(t *testing.T) {
	t.Parallel()
	a, b, clk, _ := testActivator(t, time.Minute)
	h := &hostEmptyFake{fakeBackend: b}
	a.backend = h
	b.procStuck = true
	arrangeAwake(t, a, testRoute("pending-f6.example.com"))
	h.reset() // the wake's own proof (finding 4) is not the stop's
	h.set(errors.New("the host cgroup still holds a process"))

	clk.advance(2 * time.Minute)
	a.sweepOnce()
	p := b.lastProc()
	waitFor(t, "the stuck stop", func() bool { return p.stopCalls() >= 1 })
	p.exit()
	waitFor(t, "the pending stop's proof", func() bool {
		a.sweepOnce()
		return h.count() >= 1
	})
	rep := b.lastRep()
	if rep.isClosed() {
		t.Fatal("a pending stop released the lease over a host cgroup not proven empty")
	}

	rep.setOnClose(func() {
		if n := h.passed(); n != 1 {
			t.Errorf("passing proofs before the release = %d, want 1", n)
		}
	})
	h.set(nil)
	waitFor(t, "the lease to go once the proof holds", func() bool {
		a.sweepOnce()
		return rep.isClosed() && a.entryCount() == 0
	})
}