// opset.go is the entry's operation set and its barrier (zero-downtime
// deploys spec §7.4). The current generation stays where it always was
// (e.gen, e.proc). The set holds every OTHER generation of the entry that
// is spawning, health-checking or draining, and each is tracked until its
// termination is proven. Every terminal stop joins the set before its
// final checkpoint and lease release.
package activator

import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"

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

type genRole int

const (
	roleSpawning  genRole = iota // registered; StartGen has not returned
	roleSuccessor                // spawned; health, settle or switch pending
	roleRetiring                 // was current; SIGTERM sent or about to be
)

func (r genRole) String() string {
	switch r {
	case roleSpawning:
		return "spawning"
	case roleSuccessor:
		return "successor"
	case roleRetiring:
		return "retiring"
	}
	return "unknown"
}

// genMember is one generation in an entry's operation set. Every field is
// written under a.mu. Its owner (the Swap that created it) stops it, and
// leaveSet removes it once its termination is proven. done closes at that
// moment, and that is what the barrier waits on.
type genMember struct {
	gen    Gen
	role   genRole
	proc   Proc               // nil while spawning
	cancel context.CancelFunc // cancels the attempt (spawning, successor); nil for retiring
	done   chan struct{}
	// unproven: the owner's stop could not prove termination. The member
	// stays in the set and reapMembersLocked finishes it.
	unproven  bool
	reaping   bool
	retriedAt time.Time
	// build is the worker build a shed respawn child exec'd (final review
	// F12). The generations file lists it while the member is in the set,
	// since a retained child may still run those bytes; leaveSet publishes
	// again once it has left.
	build release.StagedIdentity
}

// straySidecar is a legacy sidecar whose termination is unproven, with the
// route its proof needs (Task 7). reaping marks a re-proof in flight.
type straySidecar struct {
	proc    Proc
	rt      registry.Route
	reaping bool
}

// opSet is an entry's operation set. raised records that the parent's
// limit may be above the tier (spec §7.3): it is set before a raise is
// asked, since a raise that failed partway may still have changed the
// parent, and cleared only by a restore that succeeded. The last member to
// leave restores it; a restore that fails keeps raised, and the sweep
// retries it (retryParentRestoreLocked, final review F11). While raised is
// set with no member, no overlap is admitted. restoring marks a retry in
// flight, and restoreDone closes when it has finished: the retry is a
// lifecycle operation of the entry, joined by the barrier like a member
// (final review R2-2).
type opSet struct {
	members     map[uint64]*genMember
	raised      bool
	restoring   bool
	restoreDone chan struct{}
}

// opJoinBound bounds the barrier's wait. A member's owner needs at most a
// cancelled spawn (bounded by the control channel's deadline) plus one
// bounded stop. A var so tests can shrink it.
var opJoinBound = instancectl.DeadlineControl + actStopTimeout

// addMemberLocked registers m. The caller holds a.mu.
func (a *Service) addMemberLocked(e *actEntry, m *genMember) {
	if e.ops.members == nil {
		e.ops.members = make(map[uint64]*genMember)
	}
	e.ops.members[m.gen.N] = m
}

// isMember reports whether m is still in e's set.
func (a *Service) isMember(e *actEntry, m *genMember) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	return e != nil && e.ops.members[m.gen.N] == m
}

// joinOps is the barrier (spec §7.4). It cancels every member that is still
// an attempt (spawning or successor) and waits, bounded by opJoinBound, for
// every member to leave the set, and for a parent restore retry in flight
// to finish (final review R2-2): its write is keyed by host, so once the
// entry is gone it could lower a later entry's raised parent. It reports
// false when a member's termination was not proven, or the retry did not
// finish, in time. stop() then keeps the lease, exactly as it does for
// ErrStillRunning. An empty set returns true at once.
func (a *Service) joinOps(e *actEntry) bool {
	a.mu.Lock()
	waits := make([]chan struct{}, 0, len(e.ops.members)+1)
	for _, m := range e.ops.members {
		if m.cancel != nil {
			m.cancel()
		}
		waits = append(waits, m.done)
	}
	if e.ops.restoreDone != nil {
		waits = append(waits, e.ops.restoreDone)
	}
	hook := a.joinWaiting
	a.mu.Unlock()
	if len(waits) == 0 {
		return true
	}
	if hook != nil {
		hook(e.host, "operation set")
	}
	bound := time.NewTimer(opJoinBound)
	defer bound.Stop()
	for _, w := range waits {
		select {
		case <-w:
		case <-bound.C:
			return false
		}
	}
	return true
}

// genKiller is plan 2's generation kill (ExecBackend.KillGen and its
// SplitBackend forward): it kills what is left of generation n's own child
// cgroup and returns once that child is empty. Plan 3 owns the name; plan 5
// uses it and does not declare its own.
type genKiller interface {
	KillGen(ctx context.Context, rt registry.Route, n uint64) error
}

// hostKiller is the exec backend's host-cgroup kill: cgroup.kill on host's
// own cgroup, then wait until it is empty. It proves a worker generation
// terminated (plan 2 D4 keeps workers there).
type hostKiller interface {
	KillHost(ctx context.Context, rt registry.Route) error
}

// hostEmptier is the exec backend's host-cgroup proof for a flag-off
// terminal stop (Task 8, F6): host's own cgroup is read first and killed
// only when something is left, then waited empty. stopHostProven prefers
// it to hostKiller, so a clean stop leaves no kill mark.
type hostEmptier interface {
	EmptyHost(ctx context.Context, rt registry.Route) error
}

// stopMember stops m's process, bounded by actStopTimeout: SIGTERM, the
// generation's own graceful drain, then SIGKILL (spec §7.5 step 6). It
// reports whether termination is proven, which takes two things: the
// leader has exited, and the generation's child cgroup is empty (spec
// §7.3, §7.4). The flat childProc signals only its leader, so a process
// the app forked can outlive it in g<n>.
//
// A member with no process never got one back from its spawn. That is not
// proof: a nested start whose kill was unproven answers no process
// (startUnitAt), so its generation's child is proven empty here instead.
// Only a member that never had a number is trivially proven.
func (a *Service) stopMember(e *actEntry, m *genMember) bool {
	a.mu.Lock()
	rt := e.route
	a.mu.Unlock()
	if m.proc == nil {
		return a.handlelessGenEmpty(rt, m.gen)
	}
	var err error
	if cp, ok := m.proc.(*claimProcess); ok && m.role == roleRetiring {
		// A serving claims-on generation drains like a flat one (spec §7.5
		// step 6, index P6 #4): SIGTERM and actStopTimeout first, then
		// KillAndEmpty and the proof, with their own bound. An unready
		// candidate takes the arm below: Stop kills it at once.
		ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout+instancectl.DeadlineControl)
		err = cp.Retire(ctx, actStopTimeout)
		cancel()
	} else {
		ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		err = m.proc.Stop(ctx)
		cancel()
	}
	if errors.Is(err, ErrStillRunning) {
		return false
	}
	if err != nil {
		log.Printf("activator: stop generation %d: %v", m.gen.N, err)
	}
	return a.genTerminated(rt, m.gen, m.proc) // swapsim:M7
}

// handlelessGenEmpty is the termination proof for a generation with no
// process handle (a member that never spawned, a failed wake whose start
// returned none): its own child, proven empty through KillGen. A number of
// 0 names no child, so it is trivially proven for a member (nothing was
// ever asked to start under it). Once proven, the generation's own socket
// directory goes if it is empty (removeGenDir, rev 19 rule 3).
func (a *Service) handlelessGenEmpty(rt registry.Route, g Gen) bool {
	if g.N == 0 {
		return true
	}
	if !a.genChildEmpty(rt, &genMember{gen: g}) {
		return false
	}
	removeGenDir(rt, g)
	return true
}

// genTerminated is the ONE termination proof for a generation (spec §7.3,
// §7.4): its leader has exited AND its child is proven empty. Every path
// that would give up a generation's lease or its place in the operation
// set asks it: a member's stop, the reaper, the current generation's stop,
// a pending stop's finish, a failed wake and its retained retry. A nil
// process never ran.
//
// Once it holds for a process that ran, the generation's own socket
// directory goes if it is empty (removeGenDir, rev 19 rule 3). Its socket
// is gone by then unless the app left it behind, and the agent's control
// listener closed with the stop. A nil process proves nothing about a
// directory, so it is left; a leftover is harmless (rule 3).
func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool {
	if p == nil {
		if g.Worker {
			// A worker's host cgroup is proven whether or not a process
			// handle is left: a dead leader's descendants live there (Task 11).
			return a.genChildEmpty(rt, &genMember{gen: g})
		}
		return true
	}
	if !p.Exited() {
		return false
	}
	if !a.genChildEmpty(rt, &genMember{gen: g, proc: p}) {
		return false
	}
	removeGenDir(rt, g)
	return true
}

// genChildEmpty proves m's generation child empty, after its leader has
// exited, through plan 2's KillGen: whatever is left in g<n> (or the
// nested carlos-instance-gen@<id>:<n> unit) is killed and the child is
// waited empty. That kill is the only one that can reach a descendant the
// leader left behind, and it touches no other generation.
//
// A backend with no generation kill (a test double), or one that answers
// errNoGenContainment, has no generation children: the generation ran in
// the host cgroup, and the terminal stop's own stop evidence on that cgroup
// is the proof (plan 2, "stop evidence on the parent"). Any other error is
// an unproven termination. A retained failed wake has no such terminal stop
// behind it, so failedWakeStopped asks for the host cgroup itself
// (genChildProof's uncontained).
func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool {
	proven, _ := a.genChildProof(rt, m)
	return proven
}

// genChildProof is genChildEmpty's proof. uncontained reports that it
// proved nothing beyond the leader's exit, because the generation has no
// child of its own: the backend has no generation kill, KillGen answered
// errNoGenContainment (the flag-off ExecBackend), or it has no number (a
// flag-off worker). Such a generation ran in carlos-exec@<host>, where a
// process its leader forked can outlive it (Task 15, Astra IMPL-3 F6).
func (a *Service) genChildProof(rt registry.Route, m *genMember) (proven, uncontained bool) {
	if cp, ok := m.proc.(*claimProcess); ok {
		// A claims-on generation (Task 10): its own Stop is the proof, and
		// only a Stop that completed (KillAndEmpty, the exit, the settle and
		// the close) counts. A leader that exited after KillAndEmpty failed
		// is not terminated.
		ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
		defer cancel()
		return cp.provenStopped(ctx), false
	}
	if m.gen.Worker {
		// A worker runs in its host cgroup (plan 2 D4): that cgroup, not a
		// g<n> it never ran in, is what must be empty (spec §7.4, §7.9).
		hk, ok := a.backend.(hostKiller)
		if !ok {
			return true, false // a backend with no host cgroup: the leader's exit is all there is
		}
		ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
		defer cancel()
		if err := hk.KillHost(ctx, rt); err != nil {
			log.Printf("activator: %s: worker generation %d's host cgroup is not proven empty: %v", rt.Host, m.gen.N, err)
			return false, false
		}
		return true, false
	}
	k, ok := a.backend.(genKiller)
	if !ok || m.gen.N == 0 {
		return true, true
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	err := k.KillGen(ctx, rt, m.gen.N)
	if err == nil {
		return true, false
	}
	if errors.Is(err, errNoGenContainment) {
		return true, true
	}
	log.Printf("activator: %s: generation %d's child is not proven empty: %v", rt.Host, m.gen.N, err)
	return false, false
}

// uncontainedHostEmpty proves a failed wake's host cgroup empty when its
// generation had no child of its own (Task 15, Astra IMPL-3 F6, flag off):
// KillHost kills whatever is left in carlos-exec@<host> and waits it
// empty, as the worker proof does. An absent cgroup is empty. Only an exec
// child's wake ran there (hostCgroupIsTheWakes); a nested or unit-backed
// row's proof is its own, and a cgroup that is not the row's is never
// killed (fix round 1, ruling c).
//
// Killing is safe here because nothing of another entry can be in that
// cgroup at this point. The failed wake still holds the host's lease, and
// while it is retained the host's next wake, request or claim, is refused
// (errWakeUnproven); the failed wake never reached the legacy sidecar's
// co-spawn, which comes after every failure that reaches this cleanup;
// sidecar and seed starts run only under the host's live entry; workers
// and sidecar members are rows of their own, each with its own
// carlos-exec@<host>. What can be there is the wake's own leftovers, or a
// legacy sidecar of an earlier entry whose stop did not end it. With the
// flag off such a sidecar is untracked (straySidecars is filled only with
// generations on), and it is meant to be gone, so killing it is right. The
// killed cgroup is recreated by the next start into it (Task 19).
//
// A box with no delegated cgroup (cgroup.ErrNoDelegation) ran the wake
// unaccounted, with nothing that could reach a descendant: the leader's
// exit is all there is, as before. Keeping the lease there would refuse
// the host on this box until the agent restarts while proving nothing.
// Any other failure keeps the wake retained, and the sweep asks again.
func (a *Service) uncontainedHostEmpty(rt registry.Route, g Gen, launched Layout) bool {
	if !a.hostCgroupIsTheWakes(rt, g, launched) {
		return true // nested or unit-backed: its own proof already ran
	}
	hk, ok := a.backend.(hostKiller)
	if !ok {
		return true // a backend with no host cgroup: the leader's exit is all there is
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	err := hk.KillHost(ctx, rt)
	if err == nil {
		return true
	}
	if errors.Is(err, cgroup.ErrNoDelegation) {
		log.Printf("activator: %s: failed wake's generation %d ran with no delegated cgroup; its leader's exit is the only proof there is", rt.Host, g.N)
		return true
	}
	log.Printf("activator: %s: failed wake's generation %d had no cgroup of its own, and its host cgroup is not proven empty: %v", rt.Host, g.N, err)
	return false
}

// stopHostProven is the flag-off half of F6 for a terminal stop (Task 8):
// a flat or foreign exec row's carlos-exec@<host> is proven empty before
// the stop releases the lease. The leader's exit says nothing of a process
// it forked, and with the flag off that process sits in the host cgroup
// with no generation child to reach it. hostEmptier when the backend has
// it (read first, kill only when populated), else hostKiller; a backend
// with neither has no host cgroup, and the leader's exit is all there is.
// A claims-on process proved its own stop; a nested or unit-backed row has
// its own proof (hostCgroupIsTheWakes, on the layout the wake launched).
//
// Killing is safe here for uncontainedHostEmpty's reasons: the entry is
// stopping, its sidecar was stopped first, a supervised respawn and a
// sidecar start are joined before this runs, and nothing of another entry
// lives in that cgroup. A box with no delegated cgroup
// (cgroup.ErrNoDelegation) keeps the leader-only proof with a log line
// (ledger 127). Any other failure answers false, and the stop stays
// pending with its lease. Never called with a.mu held.
func (a *Service) stopHostProven(e *actEntry) bool {
	a.mu.Lock()
	rt, g, p, launched := e.route, e.gen, e.proc, e.launchLayout
	a.mu.Unlock()
	if _, ok := p.(*claimProcess); ok {
		return true
	}
	if !a.hostCgroupIsTheWakes(rt, g, launched) {
		return true
	}
	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	var err error
	if he, ok := a.backend.(hostEmptier); ok {
		err = he.EmptyHost(ctx, rt)
	} else if hk, ok := a.backend.(hostKiller); ok {
		err = hk.KillHost(ctx, rt)
	} else {
		return true // a backend with no host cgroup: the leader's exit is all there is
	}
	if err == nil {
		return true
	}
	if errors.Is(err, cgroup.ErrNoDelegation) {
		log.Printf("activator: %s: stopped with no delegated cgroup; its leader's exit is the only proof there is", rt.Host)
		return true
	}
	log.Printf("activator: stop %s: its host cgroup is not proven empty: %v", rt.Host, err)
	return false
}

// wakeHostProven is a flag-off wake's proof, under its lease and before
// Restore (Task 8 fix round 1, finding 4), that carlos-exec@<host> holds
// nothing: EmptyHost reads cgroup.events and kills only what is left, so a
// clean host costs one read and leaves no kill mark. Only a row whose child
// runs there is asked: an exec child (flat or foreign) or a worker, which
// spawns there whatever its layout says; a nested or unit-backed row runs
// elsewhere and has its own proofs. A backend with no hostEmptier has no
// host cgroup to read. No delegated cgroup (cgroup.ErrNoDelegation) is
// today's wake. Any other failure is errHostUnproven.
//
// Killing is safe here for uncontainedHostEmpty's reasons: the wake holds
// the host's lease, nothing of it has run yet, and the stop that released
// the lease before it proved its own processes gone, so whatever is left
// belongs to no live entry. Never called with a.mu held.
func (a *Service) wakeHostProven(ctx context.Context, e *actEntry, rt registry.Route) error {
	he, ok := a.backend.(hostEmptier)
	if !ok || rt.Backing == registry.BackingUnit {
		return nil
	}
	if !e.worker {
		if c, ok := a.backend.(Classifier); ok {
			if l, _ := c.Layout(rt); l == LayoutNested {
				return nil
			}
		}
	}
	pctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl)
	defer cancel()
	err := he.EmptyHost(pctx, rt)
	if err == nil || errors.Is(err, cgroup.ErrNoDelegation) {
		return nil
	}
	return fmt.Errorf("%w: %v", errHostUnproven, err)
}

// hostCgroupIsTheWakes reports whether a failed wake of rt may have run in
// carlos-exec@<host>, so that cgroup must be proven empty: an exec child,
// that is, a flat or foreign row. A unit-backed row runs under systemd
// (SplitBackend's unit side), and a nested row runs as a carlos-instance
// unit in its host slice; both have their own proof. A worker spawns into
// carlos-exec@<host> whatever its layout says (StartWorker), and a
// flag-off worker's wake is the only one with no number (an instance wake
// is always numbered, nextGen), so a number of 0 counts as an exec child.
// rt is the row the wake launched against, so its backing is the launch's.
//
// The layout is the one the launch used (launched), not the one the row
// reads now (final review round 1, finding 6, Astra): a flat wake whose
// layout reads fenced or conflicting by cleanup time still ran in
// carlos-exec@<host>. Only a launch that was nested skips the proof, and
// only while the row still reads nested: a launch layout that is unknown
// (a backend that cannot say, "" here), fenced or conflicting, or a
// recorded and a derived layout that disagree, requires it.
func (a *Service) hostCgroupIsTheWakes(rt registry.Route, g Gen, launched Layout) bool {
	if rt.Backing == registry.BackingUnit {
		return false
	}
	if g.N == 0 || g.Worker {
		return true
	}
	if launched != LayoutNested {
		return true
	}
	c, ok := a.backend.(Classifier)
	if !ok {
		return true
	}
	l, _ := c.Layout(rt)
	return l != LayoutNested
}

// dropMember stops m and takes it out of e's set once its termination is
// proven. It reports whether termination was proven. An unproven member
// stays in the set, marked, and the sweep's reaper finishes it.
func (a *Service) dropMember(e *actEntry, m *genMember) bool {
	proven := a.stopMember(e, m) // swapsim:M4
	if !proven {
		a.mu.Lock()
		m.unproven, m.cancel, m.retriedAt = true, nil, a.now()
		a.mu.Unlock()
		log.Printf("activator: generation %d of %s did not prove its exit; keeping it in the operation set", m.gen.N, e.host)
		return false
	}
	a.leaveSet(e, m)
	return true
}

// leaveSet removes a member whose termination is proven. When it is the
// last member and the parent was raised, the parent is restored FIRST,
// while the member still holds the set. That way no new attempt can raise
// the parent again in between (spec §7.3: the restore happens after the
// retiring child is empty). The restore is to the CURRENT generation's
// captured tier (spec §7.3): the incumbent's after a failed successor, the
// successor's after a switch. done closes last.
//
// A restore that fails keeps raised (final review F11): the member still
// leaves, since its termination is proven, but the obligation stays on the
// entry and the sweep retries it (retryParentRestoreLocked).
func (a *Service) leaveSet(e *actEntry, m *genMember) {
	a.mu.Lock()
	restore := e.ops.raised && len(e.ops.members) == 1 && e.ops.members[m.gen.N] == m
	rt, tier := e.route, e.gen.Tier
	a.mu.Unlock()
	restored := false
	if restore {
		restored = a.restoreParent(rt, tier)
	}
	a.mu.Lock()
	if restored {
		e.ops.raised = false
	}
	if e.ops.members[m.gen.N] == m {
		delete(e.ops.members, m.gen.N)
	}
	a.mu.Unlock()
	close(m.done)
	if _, listed := a.genIdentity(m.gen); listed || m.build != (release.StagedIdentity{}) {
		// Its child is proven gone: its build is no longer referenced.
		_ = a.persistGenerations()
	}
}

// restoreParent puts rt's parent back at tier, and reports whether it did.
// A backend without the swap seam has no parent to restore. Never call it
// with a.mu held.
func (a *Service) restoreParent(rt registry.Route, tier cgroup.Limit) bool {
	gb, ok := a.backend.(SwapBackend)
	if !ok {
		return true
	}
	rctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
	defer cancel()
	if err := gb.RestoreGenParent(rctx, rt, tier); err != nil && !parentUntouched(err) {
		log.Printf("activator: restore parent limit for %s: %v (kept; the sweep retries)", rt.Host, err)
		return false
	}
	return true
}

// parentUntouched reports whether a parent-limit error says no parent
// limit is in this box's hands at all: no generation containment
// (errNoGenContainment) or no delegated cgroup manager. Both are refused
// before any write, and both are fixed for the agent's life, so a raise
// they refuse changed nothing and a restore they refuse has nothing to
// put back.
func parentUntouched(err error) bool {
	return errors.Is(err, errNoGenContainment) || errors.Is(err, cgroup.ErrNoDelegation)
}

// retryParentRestoreLocked retries a parent restore that failed when the
// last member left (final review F11). It runs only while the set is empty
// and no retry is in flight; the restore itself runs off a.mu. An entry
// that is gone takes its obligation with it: the next cold spawn under an
// empty parent writes the tier afresh (cgroup.ensureParent). The caller
// holds a.mu.
//
// The retry is a lifecycle operation of the entry (final review R2-2). Only
// an awake entry that is still the host's launches one, so a stopping entry
// launches none; and a terminal stop, a hold's included, joins one in
// flight (joinOps, through restoreDone) before it releases the lease,
// forgets the entry or installs a pinned replacement. The launch and the
// state change to stopping are both under a.mu, so a stop either sees the
// retry or the retry is never launched.
func (a *Service) retryParentRestoreLocked(e *actEntry) {
	if !e.ops.raised || e.ops.restoring || len(e.ops.members) > 0 {
		return
	}
	if a.entries[e.host] != e || e.state != actAwake {
		return
	}
	done := make(chan struct{})
	e.ops.restoring, e.ops.restoreDone = true, done
	rt, tier := e.route, e.gen.Tier
	go func() {
		ok := a.restoreParent(rt, tier)
		a.mu.Lock()
		e.ops.restoring, e.ops.restoreDone = false, nil
		if ok && len(e.ops.members) == 0 {
			e.ops.raised = false
		}
		a.mu.Unlock()
		close(done)
	}()
}

// reapMembersLocked finishes members whose owner could not prove their stop.
// Once a member's process has exited (or it never had one), the reaper
// proves its generation child empty (reapExited) and only then takes it out
// of the set. Until the leader exits it re-issues Stop every
// actStopTimeout, the cadence finishPendingStops keeps for the main
// process. The caller holds a.mu.
func (a *Service) reapMembersLocked(e *actEntry) {
	for _, m := range e.ops.members {
		if !m.unproven || m.reaping {
			continue
		}
		if m.proc == nil || m.proc.Exited() {
			m.reaping = true
			go a.reapExited(e, m)
			continue
		}
		if a.now().Sub(m.retriedAt) < actStopTimeout {
			continue
		}
		m.reaping, m.retriedAt = true, a.now()
		go func(m *genMember) {
			ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
			err := m.proc.Stop(ctx)
			cancel()
			if err != nil && !errors.Is(err, ErrStillRunning) {
				log.Printf("activator: retry stop of generation %d of %s: %v", m.gen.N, e.host, err)
			}
			a.mu.Lock()
			m.reaping = false
			a.mu.Unlock()
		}(m)
	}
}

// reapExited finishes an unproven member whose leader has exited, or that
// never had one: it leaves the set only once its generation child is
// proven empty. A failed proof leaves it marked, and the next sweep tries
// the kill again.
func (a *Service) reapExited(e *actEntry, m *genMember) {
	a.mu.Lock()
	rt := e.route
	a.mu.Unlock()
	var proven bool
	if m.proc == nil {
		proven = a.handlelessGenEmpty(rt, m.gen)
	} else {
		proven = a.genTerminated(rt, m.gen, m.proc)
	}
	if proven {
		a.leaveSet(e, m)
		return
	}
	a.mu.Lock()
	m.reaping, m.retriedAt = false, a.now()
	a.mu.Unlock()
}

// The CURRENT generation is held to the same proof (spec §7.4: "each is
// tracked until its termination is proven"). stop() and a pending stop's
// finish prove its child empty before the final checkpoint and the lease
// release, and a failed wake proves its generation's child empty before it
// gives the lease back.

// releasable is THE retain-until-proven rule for an entry's lease (spec
// §7.4): a stop finishes, and its lease is released, only when every
// member has left the operation set, no parent restore retry is in flight
// (final review R2-2), no stray sidecar of the host is tracked (Task 7),
// the current generation has passed genTerminated (its leader exited and
// its child, or a worker's host cgroup, empty), and every run of the
// instance is proven stopped, its unit stopped and its cgroup empty
// (settleCmds; run spec §5.3, merge ruling M1). A run writes the database
// as the serving process does, so it is held to the same rule. It runs
// without a.mu held; genTerminated may take DeadlineControl, and a run's
// settle up to cmdStopBound.
func (a *Service) releasable(e *actEntry) bool {
	a.mu.Lock()
	members, strays, restoring := len(e.ops.members), len(a.straySidecars[e.host]), e.ops.restoring
	rt, g, p := e.route, e.gen, e.proc
	a.mu.Unlock()
	if members > 0 || strays > 0 || restoring {
		return false
	}
	return a.genTerminated(rt, g, p) && a.settleCmds(e)
}

// keepPendingLocked leaves a stop pending with its lease held: the entry
// stays in stopping, and finishPendingStops finishes it once termination
// is proven. stopRetriedAt is this attempt, so the sweep's re-stop cadence
// runs a full actStopTimeout after it. The caller holds a.mu.
func (a *Service) keepPendingLocked(e *actEntry, outcome *stopObservation) {
	e.pendingStop = true
	e.stopOutcome = outcome
	e.stopRetriedAt = a.now()
}

// reapCurrentGen finishes a pending stop whose main process has exited
// (finishPendingStops), once its termination proof holds: with generations
// on, the retain-until-proven rule (releasable); with them off, every
// sidecar the stop kept unproven re-stopped and exited (sidecarsExited, fix
// wave 1, item 1), then the host cgroup proven empty (stopHostProven, Task
// 8, F6). A failed proof leaves the stop pending, and the next sweep tries
// again. The caller set e.genReaping; this clears it. Never called with
// a.mu held.
func (a *Service) reapCurrentGen(e *actEntry) {
	proven := func(e *actEntry) bool { return a.sidecarsExited(e, true) && a.stopHostProven(e) }
	if a.gens {
		proven = a.releasable
	}
	if !proven(e) {
		a.mu.Lock()
		e.genReaping = false
		a.mu.Unlock()
		return
	}
	a.mu.Lock()
	e.genReaping, e.pendingStop = false, false
	outcome, p := e.stopOutcome, e.proc
	a.mu.Unlock()
	if outcome == nil {
		outcome = &stopObservation{}
	}
	if p == nil {
		// Only the barrier (flag on) or the host proof (flag off) left it
		// pending.
		a.finishStop(e, outcome)
		return
	}
	a.finishPendingStop(e, outcome)
}

// unprovenWake is a failed wake whose generation is not proven stopped
// (spec §7.4). Its lease (and replica, if it got one) is kept until the
// sweep proves the generation stopped; until then no new wake of the host
// starts. busy marks a retry in flight.
type unprovenWake struct {
	rt    registry.Route
	gen   Gen
	proc  Proc
	lease Lease
	rep   Rep
	busy  bool
	// build is a failed worker wake's staged identity (final review F12):
	// the generations file keeps listing it until the proof holds, since
	// the retained process may still run those bytes.
	build release.StagedIdentity
	// layout is the layout the failed wake's launch used
	// (wokenInstance.launchLayout), which every retry of the proof decides
	// by (finding 6).
	layout Layout
	// legacy: an aborted start of the host's original run unit whose
	// queued job may still run (#292, Astra P1). Proven by legacyRunGone.
	legacy bool
}

// errWakeUnproven refuses a wake while the host's last failed wake is not
// proven stopped.
var errWakeUnproven = errors.New("the last failed wake's generation is not proven stopped")

// errHostUnproven refuses a flag-off wake whose host cgroup holds processes
// it could not prove gone (wakeHostProven).
var errHostUnproven = errors.New("the host cgroup holds processes not proven gone")

// genStartUnproven reports whether a generation start's error says the
// start failed and its kill did not prove the generation gone: plan 2's
// errGenUnproven (startUnitAt) or root's instancectl.ErrGenUnproven. The
// generation may still be running, so whatever the wake holds for the host
// is kept. Both exist only under GenContainment, which can be on while
// a.gens is off, so callers never gate this on a.gens.
func genStartUnproven(err error) bool {
	return errors.Is(err, errGenUnproven) || errors.Is(err, instancectl.ErrGenUnproven)
}

// failedWakeStopped is the termination proof for a failed wake's
// generation. With a process it is genTerminated. Without one (the start
// answered none, as startUnitAt does when its kill was unproven) it is the
// generation's own child, proven empty through KillGen, and never the
// missing handle: a retained wake with no number cannot be proven at all
// and stays retained.
//
// Either way, a generation with no child of its own (flag off) is proven
// only once its host cgroup is empty too (uncontainedHostEmpty; Task 15,
// Astra IMPL-3 F6). Its leader's exit, all genTerminated can see there,
// leaves whatever the leader forked running in carlos-exec@<host>. launched
// is the layout the wake's launch used (wokenInstance.launchLayout).
//
// Once the whole proof holds, and never before, the generation's own empty
// socket directory goes (removeGenDir, rev 19 rule 3).
func (a *Service) failedWakeStopped(rt registry.Route, g Gen, p Proc, launched Layout) bool {
	if p == nil && g.N == 0 {
		log.Printf("activator: %s: failed wake left no process handle and no generation number; it cannot be proven stopped, so its lease stays held", rt.Host)
		return false
	}
	// genTerminated's proof (handlelessGenEmpty's without a process),
	// spelled out so the proof can say whether it reached a child.
	if p != nil && !p.Exited() {
		return false
	}
	proven, uncontained := a.genChildProof(rt, &genMember{gen: g, proc: p})
	if !proven {
		return false
	}
	if uncontained && !a.uncontainedHostEmpty(rt, g, launched) {
		return false
	}
	// The whole proof holds, the host cgroup's included where it is needed:
	// the generation's own socket directory goes if it is empty (rev 19
	// rule 3), as genTerminated does for every other path (Astra, review of
	// 19c07b2b: a failed wake bypassed it, and directories piled up).
	removeGenDir(rt, g)
	return true
}

// reapUnprovenWakesLocked retries each unproven failed wake's proof and,
// once it holds, gives the lease back. The caller holds a.mu.
func (a *Service) reapUnprovenWakesLocked() {
	for host, u := range a.unprovenWakes {
		if u.busy {
			continue
		}
		// A legacy start's failed wake records its outcome as its entry
		// goes, in one critical section; its proof waits for that record,
		// which it may discharge (seeded pre-land review, P2).
		if u.legacy && a.entries[host] != nil {
			continue
		}
		u.busy = true
		go a.reapUnprovenWake(host, u)
	}
}

// reapUnprovenWake is one retry of a retained failed wake. A leader that is
// still alive is stopped again, bounded. Once the generation is proven
// stopped, the replica is closed (or the lease released) and the retained
// wake is forgotten, which reopens the host to wakes.
//
// A failed close or release is logged and recorded in the stop history, and
// the wake is still forgotten: the generation, which is what retention
// guards, is proven stopped. A failed close is bounded by the lease's TTL,
// as for finishStop. A failed raw release is kept as an obligation that is
// retried with the same conditional delete (giveBack, #292).
//
// A retained legacy start (#292, Astra P1) has no process and no unit
// generation: it is proven by stopping its original run unit again
// (legacyRunGone), never by the generation proof.
func (a *Service) reapUnprovenWake(host string, u *unprovenWake) {
	if u.proc != nil && !u.proc.Exited() {
		ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		err := u.proc.Stop(ctx)
		cancel()
		if err != nil && !errors.Is(err, ErrStillRunning) {
			log.Printf("activator: %s: retry stop of failed wake's generation %d: %v", host, u.gen.N, err)
		}
	}
	proven := false
	if u.legacy {
		proven = a.legacyRunGone(u.rt) == nil
	} else {
		proven = a.failedWakeStopped(u.rt, u.gen, u.proc, u.layout)
	}
	if !proven {
		a.mu.Lock()
		u.busy = false
		a.mu.Unlock()
		return
	}
	outcome := &stopObservation{mainStop: true, mainExited: true, uncertain: true}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	var err error
	if u.rep != nil {
		err = u.rep.Close(ctx)
		outcome.replicaClosed = err == nil
	} else if u.lease != nil {
		err = releaseLease(ctx, u.lease)
		if err != nil {
			a.oweRelease(host, u.lease, err)
		}
	}
	cancel()
	outcome.leaseReleased = err == nil
	if err != nil {
		log.Printf("activator: %s: failed wake's generation %d is proven stopped, but giving its lease back failed: %v", host, u.gen.N, err)
	} else {
		log.Printf("activator: %s: failed wake's generation %d is proven stopped; lease given back", host, u.gen.N)
	}
	a.mu.Lock()
	if h := a.stopHistory[host]; u.legacy && h != nil && h.uncertain && h.owedBy == u {
		// The history's only uncertainty was this start's, and #292's
		// whole proof now holds: it is discharged, and the stages the
		// failed wake never reached hold (seeded pre-land review, P2).
		outcome.uncertain = false
		outcome.sidecarStop, outcome.sidecarExited = true, true
		if u.rep == nil {
			outcome.replicaClosed = true
		}
		a.stopHistory[host] = outcome
	} else {
		a.rememberStopLocked(host, outcome)
	}
	if a.unprovenWakes[host] == u {
		delete(a.unprovenWakes, host)
	}
	u.busy = false
	a.mu.Unlock()
	if _, listed := a.genIdentity(u.gen); listed || u.build != (release.StagedIdentity{}) {
		// Proven gone: its build is no longer referenced.
		_ = a.persistGenerations()
	}
}