package activator

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"

	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

var ErrAttemptOver = moves.ErrAttemptOver

type StartPolicy struct {
	// RecoveryFence is the owning move cut generation, supplied by its adapter.
	RecoveryFence uint64
	MaxStarts     int
	Window        time.Duration
	Deadline      time.Time
	Probe         bool
	Prepared      *moves.InstallManifest
	Inputs        *provision.MoveInputs
	// Pin is a hold swap's chosen build (zero-downtime spec §7.8). Prepare
	// must build a start that execs exactly Pin.StagedPath after checking
	// Pin.Digest. startAdmitted refuses a ClaimStart whose Version is not
	// Pin.Build.
	Pin *Gen
	// Gen is the generation the activator chose for this start: its number
	// and the sockets the provider must bind (zero-downtime plan 3, Task
	// 10). Zero is today's start on the canonical socket. startAdmitted
	// refuses a ClaimStart that bound anything else (genMatches).
	Gen Gen
	// Fallback is a first wake's good build (zero-downtime spec §8.2), set
	// only beside Pin, by plan 5's planner. It is honoured once the isolated
	// launch opens; on this base the launch refuses before anything runs
	// (stepStale), so there is nothing to fall back from. Any fallback arm
	// added later must record its build first (recordBuild, settings spec
	// §4.2; ruling R-EXEC-CLAIMFALLBACK).
	Fallback *Gen
	// HealthBy, when set, caps the pinned start's Ready window: a first
	// wake's health check ends at the earlier of Window and HealthBy (spec
	// §8.2: the waiters' deadline minus 25 s).
	HealthBy time.Time
	// Gate, when set, is the attempt gate (taking a.mu itself): a claims-on
	// hold's pinCheck, or a planned wake's ownership check (plan 5).
	// startAdmitted runs it in the launch's before hook, immediately before
	// every exec, after the admission and Restore (spec §6.3, §7.4).
	Gate func() error
	// PinCheck, when set, is the planner's selection gate (plan 5's
	// WakePlan.Check) for the pin about to exec. startAdmitted runs it
	// beside Gate, at every exec.
	PinCheck func(pin Gen) error
	// Held is a held instance's target (settings spec §4.3 rule 3), "" when
	// the wake is not held. startAdmitted's before hook refuses a start
	// whose hashed build is neither the target nor a planned wake's
	// recovery boot (heldIdentityOK), before the record and the exec. The
	// activator sets it; providers ignore it.
	Held string
}
type startPolicy = StartPolicy

// ClaimRuntime is assembled by the host owner before Service.Start.
// Admit reserves the first serving attempt and binds its resulting cold session.
// Prepared probes instead consume the moves owner's single-use admission.
type ClaimRuntime struct {
	// ResumedAdmission verifies a journalled recovery that persisted prelaunch
	// under the existing session during this admission call.
	ResumedAdmission      func(context.Context, hostclaim.Observed, hostclaim.Observed) error
	RecoveryAttemptsReady func() bool
	FailedAdmission       func(context.Context, hostclaim.Observed, error) error
	Box                   string
	// Bucket and LeaseRegion are the claim actor's bucket and the region its
	// leases sign for: a claimed wake leases Holder.Location there, and the
	// instance's build-state.json sits beside that lease (settings spec
	// §4.2, ruling R-CLAIMPREFIX).
	Bucket, LeaseRegion string
	Read                func(context.Context, string) (hostclaim.Observed, error)
	StopForeign         func(context.Context, string) error
	Prerequisites       func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
	Admit               func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
	Prepare             func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
	// Pins reports that Prepare honours StartPolicy.Pin. A hold swap is
	// refused on a claims-on box whose runtime does not, before anything
	// stops.
	Pins bool
	// Generations reports that Prepare honours StartPolicy.Gen (binds its
	// sockets and reports them in ClaimStart.Gen). Without it a claims-on
	// box keeps one generation per entry, on the canonical socket, and
	// every swap is refused before anything is raised, stopped or spawned
	// (errClaimsNoGenerations).
	Generations bool
}

// ClaimStart is one admitted session. Its closures retain the admission tuple.
// Only Restore may purge data. Process retries reuse it and the live replication.
type ClaimStart struct {
	Live           context.Context
	PrepareLaunch  func(context.Context) (*RoleLaunch, error)
	CurrentAttempt func() hostclaim.WakeRetryAttempt
	RoleLauncher   *RoleLaunch
	// OpenProbe journals the first app-start time before execution and returns
	// that same time on retries, including after controller restart. Its
	// provider binds the moves attempt and admission, never resets Opened.
	OpenProbe       func(context.Context, time.Time) (time.Time, error)
	Work            hostwrite.Work
	Identity        hostwrite.Identity
	Manager         *cgroup.Manager
	Host            string
	Limit           cgroup.Limit
	Restore         func(context.Context) error
	Materialize     func(context.Context) error
	RetryClaimWrite func(hostclaim.Observed, hostclaim.Observed) error
	Build           func() *exec.Cmd
	Ready           func(context.Context, *exec.Cmd) error
	// Reserve(0) validates the already reserved first attempt. Later calls CAS.
	Reserve   func(context.Context, int) error
	Failed    func(context.Context, int) error
	Healthy   func(context.Context, int, *exec.Cmd) error
	Replicate func(context.Context, string) (Rep, error)
	Publish   func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)
	// Retain owns uncertain process or replica cleanup until it can be proved.
	Retain  func(Proc, Rep, error)
	Version string
	// Staged is the file the launch hashed, from one resolution of the
	// host's live link (SnapshotClaimLaunch): the version it is staged
	// under, its path and its digest (ExecutableSHA256). Version is a
	// separate read of the link, so a repoint between the two can make them
	// disagree; startAdmitted records Staged.Version, and refuses a start
	// whose Staged.Version is not Version (settings spec §4.2, ruling
	// R-EXEC-CLAIMVERSION).
	Staged release.StagedIdentity
	// Gen is what the provider bound. startAdmitted refuses a start whose
	// number, socket or control socket is not StartPolicy.Gen's.
	Gen Gen
}

// genMatches reports whether a provider bound exactly the generation the
// activator chose: its number, its socket and its control socket. A handle
// that names a socket nobody serves would route requests nowhere.
func genMatches(want, got Gen) bool {
	return want.N == got.N && want.Socket == got.Socket && want.Control == got.Control
}

func (s ClaimStart) valid() bool {
	return s.Work != nil && s.Identity.Config != "" && s.Identity.Release != "" && s.Host != "" &&
		s.Restore != nil && s.Materialize != nil && s.Build != nil && s.Ready != nil && s.Reserve != nil &&
		(s.RoleLauncher != nil || s.PrepareLaunch != nil) && s.CurrentAttempt != nil && s.Failed != nil && s.Healthy != nil && s.Replicate != nil && s.Publish != nil && s.Retain != nil
}

func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, error) {
	r := a.cfg.Claims
	if r == nil || r.Box == "" || r.Read == nil || r.StopForeign == nil {
		return nil, hostwrite.ErrUnavailable
	}
	// A failed wake whose generation is not proven stopped still holds the
	// host, and its process may still run against the database
	// (zero-downtime spec §7.4). As doWake does, no new wake starts until
	// the sweep proves it, and a stray legacy sidecar refuses it the same
	// way.
	a.mu.Lock()
	_, unproven := a.unprovenWakes[e.host]
	unproven = unproven || len(a.straySidecars[e.host]) > 0
	a.mu.Unlock()
	if unproven {
		return nil, fmt.Errorf("%s: refusing to wake: %w", e.host, errWakeUnproven)
	}
	observed, err := r.Read(ctx, e.host)
	if err != nil {
		return nil, err
	}
	// K6 precedes registry reads, lease acquisition, materialization and policy checks.
	executor := ""
	if observed.Claim != nil {
		executor = observed.Claim.Op.Exec
		if observed.Claim.Op.Kind == hostclaim.KindServing && observed.Claim.Holder != nil {
			executor = observed.Claim.Holder.WriterBox
		}
	}
	if observed.Claim == nil || executor != r.Box {
		stopErr := r.StopForeign(ctx, e.host)
		return nil, probeRefused(e, errors.Join(hostbox.ErrChanged, stopErr))
	}
	if e.route.WakeIncarnation != "" && (observed.Claim == nil || observed.Claim.Holder == nil || observed.Claim.Holder.Incarnation != e.route.WakeIncarnation) {
		return nil, probeRefused(e, hostbox.ErrChanged)
	}
	// A probe (ProbeWake, moves Plan 3 Task 5a) is admitted only at its
	// op's probing phase (K6's probe row) and runs under its own policy.
	probe := e.probe
	if err := hostbox.WakePhase(observed.Claim, r.Box, probe != nil || e.recoveryPolicy != nil); err != nil {
		return nil, probeRefused(e, err)
	}
	if probe != nil && observed.Claim.Op.ID != probe.op {
		return nil, probeRefused(e, errProbeOp)
	}
	if e.recoveryPolicy != nil && (!e.recoveryPolicy.Probe || !recoveryProbeKind(observed.Claim.Op.Kind)) {
		return nil, hostbox.ErrChanged
	}
	if r.Prerequisites == nil || r.Admit == nil || r.Prepare == nil || r.FailedAdmission == nil {
		return nil, hostwrite.ErrUnavailable
	}
	rt := e.route
	if a.Routes != nil {
		fresh, err := a.Routes.Get(e.host)
		if err != nil {
			return nil, err
		}
		if fresh == nil {
			return nil, hostbox.ErrChanged
		}
		rt = *fresh
	}
	// The held wake check (settings spec §4.3 rule 3), as doWake's step
	// 0.65, before planning and before the admission. Never for a hold's
	// pinned wake (e.pinCheck: the swap chose its build; planned claimed
	// wakes set e.pin too, below, so e.pin alone does not say "hold"), and
	// never for a moves probe, which runs the landing's prepared build under
	// its own policy. The launcher execs the claim inputs' file, so a
	// retained copy cannot be injected here (direct false): a target not
	// installed is asked of root and waited for, and the row re-read. The
	// launch's before hook refuses a hashed build that is not the target
	// (StartPolicy.Held).
	var held string
	if e.pinCheck == nil && probe == nil && e.recoveryPolicy == nil {
		held, _, err = a.wakeCheck(ctx, rt.Host, false)
		if err != nil {
			return nil, fmt.Errorf("%s: %w", rt.Host, err)
		}
		if held != "" && a.Routes != nil {
			fresh, err := a.Routes.Get(e.host)
			if err != nil {
				return nil, err
			}
			if fresh == nil {
				return nil, hostbox.ErrChanged
			}
			rt = *fresh
		}
	}
	// Claimed wakes run plan 3's pinned start (StartPolicy.Pin) and get the
	// same first-wake budget and fallback as an unclaimed wake (zero-downtime
	// spec §8.2): firstWake extends the waiters' deadline in begin, and the
	// policy below carries the health cap and the good build. The planner
	// is called off a.mu. On this base every claims-on launch refuses
	// before its exec hook (RoleLaunch.Start), so none of it runs a build.
	// A moves probe (ProbeWake) is never planned: it runs the landing's
	// prepared build under its own policy, and a plan's outcome must not be
	// recorded against a build it did not boot.
	if a.Planner != nil && e.pin == nil && probe == nil {
		// As doWake: an earlier planned wake's outcome is delivered before
		// this one plans (final review F3), and a wait that ends first
		// refuses the wake (final review wave 2, N1).
		if err := a.awaitOutcomes(ctx, rt.Host); err != nil {
			return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err)
		}
		plan := a.Planner(ctx, rt)
		if plan.Refuse != "" {
			return nil, fmt.Errorf("%s: refusing to wake: %s", rt.Host, plan.Refuse)
		}
		if plan.planned() {
			pin := genOf(plan.Boot)
			a.mu.Lock()
			e.pin, e.plan, e.firstWake = &pin, plan, plan.FirstWake
			a.mu.Unlock()
		}
	}
	// A pinned or planned build is published before the admission, as
	// doWake does before its lease (spec §4.2, §8.5): claimed wakes never
	// reach doWake's publication. A failed write refuses the wake with
	// nothing admitted and no build blamed.
	a.mu.Lock()
	staged := e.pin != nil || e.plan.planned()
	a.mu.Unlock()
	if staged {
		if err := a.publishBeforeExec(); err != nil {
			return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err)
		}
	}
	policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin, Held: held}
	if e.recoveryPolicy != nil {
		policy = *e.recoveryPolicy
	} else if probe != nil {
		// K1: the row is still held by the probe's hold op.
		if !probeHeld(rt, probe.holdOp) {
			return nil, probeRefused(e, errProbeRow)
		}
		policy = probe.policy
	} else if e.pin != nil && e.pinCheck != nil {
		// A claims-on hold's pinned wake: the attempt gate before anything
		// is admitted, and again in the launch's before hook at every exec,
		// after the admission and Restore (spec §6.3, §7.4), as
		// startPinnedChecked does for a claims-off one.
		gate := func() error {
			a.mu.Lock()
			defer a.mu.Unlock()
			return e.pinCheck()
		}
		if err := gate(); err != nil {
			return nil, err
		}
		policy.Gate = gate
	}
	if a.gens && r.Generations {
		policy.Gen = a.claimGenAt(rt, nextGen())
	}
	if p := e.plan; p.planned() && p.FirstWake && p.Mode == ModeOverlap && p.Fallback != nil {
		// The same budget as spawnPlanned: B's health until the waiters'
		// deadline minus 25 s, then the good build in its place. Fallback is
		// declared on StartPolicy but honoured only once the isolated launch
		// opens (startAdmitted does not run it on this base).
		fb := genOf(*p.Fallback)
		policy.Fallback = &fb
		policy.HealthBy = e.wakeStart.Add(firstWakeBudget - firstWakeKillReserve)
	}
	if p := e.plan; p.planned() {
		// At every claimed planned exec, B and its fallback alike, inside
		// the work after the admission and Restore (startAdmitted's
		// beforeStart): the ownership check (spec §7.4), and the planner's
		// selection gate for the pin about to exec (§8.1–8.4; P5 review
		// finding 3). A refusal there records nothing and starts nothing;
		// the wake fails and the next one is planned afresh. Both run off
		// a.mu (the gate takes it itself).
		if policy.Gate == nil {
			policy.Gate = func() error {
				a.mu.Lock()
				defer a.mu.Unlock()
				return a.ownedLocked(e, true)
			}
		}
		if p.Check != nil {
			policy.PinCheck = func(pin Gen) error {
				if err := p.Check(pin.Build); err != nil {
					return fmt.Errorf("%w: %v", errPlanStale, err)
				}
				return nil
			}
		}
	}
	if err := r.Prerequisites(ctx, rt, observed, policy); err != nil {
		return nil, err
	}
	admitted, err := r.Admit(ctx, rt, observed, policy)
	if err != nil {
		return nil, err
	}
	if admitted.Claim == nil || admitted.Claim.Holder == nil || observed.Claim.Holder == nil || admitted.Claim.Holder.Incarnation != observed.Claim.Holder.Incarnation {
		return nil, hostbox.ErrChanged
	}
	if admitted.Claim.Holder.Session <= observed.Claim.Holder.Session || admitted.Claim.Holder.Generation == observed.Claim.Holder.Generation {
		if !recoveryProbeKind(observed.Claim.Op.Kind) || !hostclaim.SameWriterIdentity(observed, admitted) || r.ResumedAdmission == nil {
			return nil, hostbox.ErrChanged
		}
		if err := r.ResumedAdmission(ctx, observed, admitted); err != nil {
			return nil, err
		}
	}
	e.route = rt
	if e.recoveryPolicy != nil {
		e.probeAdmitted = admitted
	}
	if err := a.startAdmitted(ctx, e, admitted, policy); err != nil {
		recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancelRecovery()
		err = errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err))
		if probe != nil {
			err = errors.Join(ErrProbeFailed, err)
		}
		// claimedPlanResult carries a planned wake's outcome to afterWake: a
		// non-nil result with an error carries no stopOutcome, so wake()
		// records the uncertain stop it records today. A probe is never
		// planned, so its result stays nil as on main.
		return a.claimedPlanResult(e, nil), err
	}
	if probe != nil {
		probe.admitted = &admitted
	}
	return a.claimedPlanResult(e, e.claimResult), nil
}

// claimStartBackoff is startAdmitted's wait before its second start; the
// wait doubles before each start after that (1, 2, 4, 8 s). A var so tests
// can shrink it.
var claimStartBackoff = time.Second

func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostclaim.Observed, policy startPolicy) (result error) {
	r := a.cfg.Claims
	if !a.cfg.HostWrites.Enabled || r == nil || r.Read == nil || r.Prepare == nil || e == nil || admitted.Host != e.host || policy.MaxStarts < 1 || policy.MaxStarts > 5 || policy.Window <= 0 || policy.Window > 30*time.Second {
		return hostwrite.ErrUnavailable
	}
	budget := &probeWindow{policy: policy, now: time.Now}
	defer func() { result = budget.result(result) }()
	if !policy.Deadline.IsZero() {
		bounded, cancel := context.WithDeadline(ctx, policy.Deadline)
		defer cancel()
		ctx = bounded
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	if err := hostbox.WakePhase(admitted.Claim, r.Box, policy.Probe); err != nil {
		return err
	}
	fresh, err := r.Read(ctx, e.host)
	if err != nil {
		return err
	}
	if fresh.ETag != admitted.ETag && !(recoveryProbeKind(admitted.Claim.Op.Kind) && hostclaim.SameWriterIdentity(fresh, admitted)) {
		return hostbox.ErrChanged
	}
	s, err := r.Prepare(ctx, e.route, admitted, policy)
	if err != nil {
		return err
	}
	if !s.valid() || s.Host != e.host || (policy.Probe && s.OpenProbe == nil) {
		return hostwrite.ErrUnavailable
	}
	if policy.Pin != nil && s.Version != policy.Pin.Build {
		// A hold swap's wake must run the build its attempt chose
		// (zero-downtime spec §7.8). A start prepared for anything else is
		// refused before Restore touches the database.
		return hostwrite.ErrUnavailable
	}
	if err := a.claimedBuild(e.host, s); err != nil {
		// The bytes the snapshot hashed are not the build the start names:
		// nothing could record the build that would run. Refused before
		// Restore touches the database, and no failed attempt of the build.
		return err
	}
	if policy.Gen.N != 0 && !genMatches(policy.Gen, s.Gen) {
		// The provider bound something other than the generation the
		// activator chose: its handle would name a socket nobody serves.
		// Refused before Restore touches the database.
		return hostwrite.ErrUnavailable
	}
	if s.Live != nil {
		if err := s.Live.Err(); err != nil {
			return err
		}
		joined, cancel := context.WithCancel(ctx)
		defer cancel()
		stop := context.AfterFunc(s.Live, cancel)
		defer stop()
		ctx = joined
	}
	prefix, err := hostbox.GenerationPrefix(*admitted.Claim.Holder)
	if err != nil {
		return err
	}
	if err := s.Restore(ctx); err != nil {
		return err
	}
	var rep Rep
	var retained bool
	defer func() {
		if result == nil || rep == nil || retained {
			return
		}
		cleanup, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := rep.Close(cleanup); err != nil {
			s.Retain(nil, rep, err)
			result = errors.Join(result, err)
		}
	}()
	var last error
	for attempt := 0; attempt < policy.MaxStarts; attempt++ {
		if err := budget.result(ctx.Err()); err != nil {
			return err
		}
		if err := s.Reserve(ctx, attempt); err != nil {
			return err
		}
		if err := s.Materialize(ctx); err != nil {
			return errors.Join(err, s.Failed(ctx, attempt))
		}
		if s.RoleLauncher == nil {
			s.RoleLauncher, err = s.PrepareLaunch(ctx)
			if err != nil {
				return errors.Join(err, s.Failed(ctx, attempt))
			}
			if s.RoleLauncher == nil {
				return errors.Join(hostwrite.ErrUnavailable, s.Failed(ctx, attempt))
			}
		}
		// This hook runs after strict preparation and immediately before Start.
		// Restore and materialization consume only the separate cut deadline.
		reached := false // the exec hook ran: an error before it is the admission's (claimAdmission)
		unrecorded := false
		beforeStart := func() error {
			reached = true
			if policy.Gate != nil {
				// The attempt gate at the exec: a claims-on hold's pinCheck,
				// or a planned wake's ownership check (plan 5).
				if err := policy.Gate(); err != nil {
					return err
				}
			}
			if policy.PinCheck != nil && policy.Pin != nil {
				// The planner's selection gate for the build about to exec
				// (plan 5): a failure recorded since planning holds it.
				if err := policy.PinCheck(*policy.Pin); err != nil {
					return err
				}
			}
			// The build state names the build before it runs against the
			// database (settings spec §4.2), beside the claimed lease's
			// lock.json: the claim actor's bucket at Holder.Location, never
			// the session's g-<gen>/ prefix (ruling R-CLAIMPREFIX). The file
			// is already open and verified against its digest here
			// (RoleLaunch.Start). Every exec records, a retry's included;
			// the same build makes the write idempotent. Moves probes come
			// through here too (ruling R-EXEC-PROBES).
			//
			// A held wake first checks the build the snapshot hashed
			// (s.Staged, the version of the bytes opened; Task 5d) against
			// its target (settings spec §4.3 rule 3, Astra I2 #4): root may
			// have repointed between the wait and the snapshot. A refusal
			// records nothing, and like a refused record it is terminal for
			// this loop: no failed attempt, no retry (unrecorded).
			if policy.Held != "" {
				a.mu.Lock()
				plan, route := e.plan, e.route
				a.mu.Unlock()
				if !heldIdentityOK(policy.Held, s.Staged, plan, route) {
					unrecorded = true
					log.Printf("activator: %s not started: it is held on %s, and the launch hashed %q (the live link moved after the check); the next wake checks again", e.host, policy.Held, s.Staged.Version)
					return fmt.Errorf("%w: %s would run %q, wants %s", errStaleLink, e.host, s.Staged.Version, policy.Held)
				}
			}
			at := WithBuildStateLocation(ctx, BuildStateLocation{Bucket: r.Bucket, Prefix: admitted.Claim.Holder.Location})
			// A planned recovery boot records with the planner's selection
			// (ruling R-PLANNED-RECOVERY), as doWake's step 2b does.
			a.mu.Lock()
			plan, route := e.plan, e.route
			a.mu.Unlock()
			at = withRecoverySelection(at, plan, route, s.Staged.Version)
			if err := a.recordBuild(at, e.host, s.Staged.Version); err != nil {
				unrecorded = true
				return err
			}
			if policy.Gate != nil {
				// Asked again after the record, as startSidecar does: a
				// fence raised or a lease lost while the build state was
				// written (the PUT) execs nothing (Task 12a).
				if err := policy.Gate(); err != nil {
					return err
				}
			}
			if err := budget.start(ctx, s.OpenProbe); err != nil {
				return err
			}
			if e.pin != nil {
				// A hold's pinned wake has reached its build's exec: from
				// here a failure is the build's (Task 6, markPinExec).
				a.markPinExec(e)
			}
			return nil
		}
		cmd, prepared, settled, producer, startErr := s.RoleLauncher.Start(ctx, "main", s.CurrentAttempt(), s.Build, beforeStart)
		// A refusal that came back before the exec hook (the closed isolated
		// launch, a claim read, an absent claim, a changed stamp) ran nothing
		// of the build: it is the admission's, never the build's failure.
		startErr = claimAdmission(startErr, reached)

		p := newClaimProcessFor(cmd, prepared, s.RoleLauncher.Root != nil)
		p.settle = settled
		p.bindProducer(producer)
		if unrecorded {
			// A refused record, or a held wake's refused identity, ran
			// nothing of the build (ruling R-EXEC-CLAIMRETRY): no failed
			// attempt is written (s.Failed), and the loop does not retry it.
			// The wake fails as fenced (memberSafeWakeReason).
			cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second)
			cleanErr := p.Stop(cleanup)
			cancelCleanup()
			if cleanErr != nil {
				s.Retain(p, rep, errors.Join(startErr, cleanErr))
				retained = true
				return errors.Join(startErr, cleanErr)
			}
			return startErr
		}
		if startErr == nil {
			deadline := time.Now().Add(policy.Window)
			if policy.Probe {
				deadline = budget.opened.Add(policy.Window)
			}
			if !policy.HealthBy.IsZero() && policy.HealthBy.Before(deadline) {
				// A first wake's pinned build: health until the waiters'
				// deadline minus 25 s (zero-downtime spec §8.2).
				deadline = policy.HealthBy
			}
			readyCtx, cancelReady := context.WithDeadline(ctx, deadline)
			startErr = s.Ready(readyCtx, cmd)
			cancelReady()
			startErr = budget.result(startErr)
		}
		ready := startErr == nil
		budget.ready = ready
		if ready {
			rep, startErr = s.Replicate(ctx, prefix)
			if startErr == nil && rep == nil {
				startErr = hostwrite.ErrUnavailable
			}
		}
		if startErr == nil {
			current, err := r.Read(ctx, e.host)
			if err != nil {
				startErr = err
			} else if !sameAdmission(current, admitted) {
				startErr = hostbox.ErrChanged
			} else {
				_, startErr = s.Publish(ctx, current)
			}
		}
		if startErr == nil {
			startErr = s.Healthy(ctx, attempt, cmd)
		}
		if startErr == nil {
			a.mu.Lock()
			if e.pinFailed != nil && policy.Pin != nil && e.pinFailed.Build == policy.Pin.Build {
				e.pinFailed, e.pinFailReason = nil, "" // a retry of the same pin served
			}
			a.mu.Unlock()
			// The generation's handle: the sockets the provider bound
			// (genMatches, above). Zero with generations off, and wake()
			// falls back to the canonical handle as before.
			g := policy.Gen
			if g.N != 0 {
				g.Build = s.Version
				if policy.Pin != nil {
					g.StagedPath, g.Digest, g.Demand = policy.Pin.StagedPath, policy.Pin.Digest, policy.Pin.Demand
				}
			}
			if (policy.Pin == nil || g.N == 0) && s.Staged.StagedPath != "" && s.Staged.Digest != "" && s.Staged.Version == s.Version {
				// The file this start ran (the hashed resolution of the
				// live link), so a legacy sidecar restart runs that file
				// by path, never whatever the link names by then (Task
				// 5d; ruling R-EXEC-SIDECAR). With no number, wake()
				// keeps it on the canonical handle it gives this start.
				// A pinned start with no number takes the same file: its
				// version was checked against Pin.Build above, and
				// s.Staged is what ran; the demand is the pin's (Task 12a).
				g.Build, g.StagedPath, g.Digest = s.Version, s.Staged.StagedPath, s.Staged.Digest
				if policy.Pin != nil {
					g.Demand = policy.Pin.Demand
				}
			}
			e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, gen: g, stopOutcome: &stopObservation{}}
			return nil
		}
		cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second)
		cleanErr := p.Stop(cleanup)
		cancelCleanup()
		if cleanErr != nil {
			s.Retain(p, rep, errors.Join(startErr, cleanErr))
			retained = true
			return errors.Join(startErr, cleanErr)
		}
		if err := s.Failed(ctx, attempt); err != nil {
			return errors.Join(startErr, err)
		}
		last = startErr
		// THE outcome classifier (Task 4): a start the wake's lifecycle ended
		// (maintenance, Shutdown, a stop), that the attempt gate refused, or
		// that the launch refused before its exec hook is no failure of the
		// build and records nothing (spec §6.4, §7.4). Only a pinned build
		// that ran (cmd) and failed before it was ready is recorded, for plan
		// 5's events (a failed first wake, spec §8.2).
		if policy.Pin != nil && !ready && cmd != nil && classifyStep(ctx, startErr) == stepFailed {
			a.mu.Lock()
			if e.pinFailed == nil {
				failed := *policy.Pin
				e.pinFailed, e.pinFailReason = &failed, startErr.Error()
			}
			a.mu.Unlock()
		}
		// A replication/publication failure has an uncertain remote outcome.
		// The next process start cannot reuse that opening as another attempt.
		if ready {
			return last
		}
		if errors.Is(startErr, ErrAttemptOver) {
			return startErr
		}
		if attempt+1 < policy.MaxStarts {
			waitCtx := ctx
			cancelWait := func() {}
			if policy.Probe && !budget.opened.IsZero() {
				waitCtx, cancelWait = context.WithDeadline(ctx, budget.opened.Add(policy.Window))
			}
			timer := time.NewTimer(claimStartBackoff << attempt)
			select {
			case <-waitCtx.Done():
				timer.Stop()
				cancelWait()
				return budget.result(waitCtx.Err())
			case <-timer.C:
				cancelWait()
			}
		}
	}
	return last
}

type claimContainment interface {
	KillAndEmpty(context.Context) error
	Close() error
}

type claimProcess struct {
	cmd             *exec.Cmd
	prepared        claimContainment
	done            chan struct{}
	mu              sync.Mutex
	drained, closed bool
	// root marks a role root launched in an isolated unit: the agent holds
	// no containment, settle is root's stop of the reservation, and done
	// follows the unit's main process through a pidfd.
	root           bool
	settle         func(context.Context) error
	producerFinish func(error)
}

func newClaimProcess(cmd *exec.Cmd, p *cgroup.Prepared) *claimProcess {
	c := &claimProcess{cmd: cmd, done: make(chan struct{})}
	if p != nil {
		c.prepared = p
	}
	if cmd == nil || cmd.Process == nil {
		close(c.done)
	} else {
		go func() { _ = cmd.Wait(); close(c.done) }()
	}
	return c
}
func (p *claimProcess) Exited() bool {
	select {
	case <-p.done:
		return true
	default:
		return false
	}
}
func (p *claimProcess) Stop(ctx context.Context) error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.closed {
		return nil
	}
	if p.root {
		return p.stopRoot(ctx)
	}
	if p.prepared == nil {
		if p.cmd != nil && p.cmd.Process != nil {
			return hostwrite.ErrUnavailable
		}
		p.closed = true
		if p.producerFinish != nil {
			p.producerFinish(nil)
		}
		return nil
	}
	if !p.drained {
		if err := p.prepared.KillAndEmpty(ctx); err != nil {
			return err
		}
		p.drained = true
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-p.done:
	}
	if p.settle != nil {
		if err := p.settle(ctx); err != nil {
			return err
		}
	}
	if err := p.prepared.Close(); err != nil {
		return err
	}
	p.closed = true
	if p.producerFinish != nil {
		p.producerFinish(nil)
	}
	return nil
}

// Retire is a serving claims-on generation's retirement (zero-downtime spec
// §7.5 step 6, index P6 #4): SIGTERM to its leader, the app's own drain for
// up to grace, then Stop, whose KillAndEmpty on the generation's own child
// and the emptiness proof are unchanged. A candidate that never became
// ready is not retired: its owner Stops it, which kills at once.
func (p *claimProcess) Retire(ctx context.Context, grace time.Duration) error {
	if p.cmd != nil && p.cmd.Process != nil && !p.Exited() {
		err := p.cmd.Process.Signal(syscall.SIGTERM)
		switch {
		case err == nil, errors.Is(err, os.ErrProcessDone):
			t := time.NewTimer(grace)
			select {
			case <-p.done:
			case <-t.C:
			case <-ctx.Done():
			}
			t.Stop()
		default:
			// No drain is possible; Stop still kills and proves it.
			log.Printf("activator: SIGTERM to a retiring claims-on generation (pid %d): %v; stopping it without a drain", p.cmd.Process.Pid, err)
		}
	}
	return p.Stop(ctx)
}

// provenStopped is a claims-on generation's termination proof (zero-downtime
// spec §7.4): its Stop has completed, which takes KillAndEmpty on its own
// child, the exit, the settle and the close. Stop is idempotent once
// closed, so this retries a Stop that failed before.
func (p *claimProcess) provenStopped(ctx context.Context) bool {
	if err := p.Stop(ctx); err != nil {
		return false
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.closed
}

func sameAdmission(a, b hostclaim.Observed) bool {
	if a.Absent || b.Absent || a.Host != b.Host || a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil {
		return false
	}
	x, y := a.Claim, b.Claim
	return x.Holder.Incarnation == y.Holder.Incarnation && x.Holder.Stamp == y.Holder.Stamp && x.Holder.Session == y.Holder.Session && x.Holder.Generation == y.Holder.Generation && x.Op.ID == y.Op.ID && x.Op.Kind == y.Op.Kind && x.Op.Phase == y.Op.Phase && x.Op.Exec == y.Op.Exec
}

func (p *claimProcess) bindProducer(producer RoleProducer) {
	if producer.Context == nil || producer.Finish == nil {
		return
	}
	p.producerFinish = producer.Finish
	go func() {
		select {
		case <-producer.Context.Done():
		case <-p.done:
		}
		cleanup, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		for {
			if err := p.Stop(cleanup); err == nil {
				return
			}
			select {
			case <-cleanup.Done():
				return
			case <-time.After(100 * time.Millisecond):
			}
		}
	}()
}