package host

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"os/exec"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/stopevidence"
)

// ClaimLaunchInputs is an in-process adapter result, never a new persisted
// held-input format. The moves adapter retains its MoveInputs and manifest.
type ClaimLaunchInputs struct {
	Unit             *hostbox.UnitLaunchInputs
	claimWrite       func(hostclaim.Observed, hostclaim.Observed) error
	Roles            []string
	Identity         hostwrite.Identity
	Baseline         *genselect.Plan
	Admission        hostclaim.Admission
	Executable       string
	ExecutableSHA256 string
	Args, Env        []string
	DBPath, Version  string
	// Staged is the file ExecutableSHA256 hashes and the version it is
	// staged under (ClaimServingLaunch.Staged); Prepare hands it to the
	// start, which records it before the exec (settings spec §4.2).
	Staged   release.StagedIdentity
	Limit    cgroup.Limit
	Empty    bool
	Prepared bool
	// ReadyPath is the probe's declared readiness path (moves §7.7,
	// moves.ready_path). Serving and committed wakes ignore it.
	ReadyPath string
}

type ClaimWakeInputs interface {
	// Select is read-only. It verifies the exact ordinary or held input set.
	Select(context.Context, registry.Route, hostclaim.Observed, activator.StartPolicy) (ClaimLaunchInputs, error)
	// Materialize invokes captured RenderHeld/MaterializeHeld when held and
	// establishes the origin-qualified files manifest before execution.
	Materialize(context.Context, hostclaim.Observed, ClaimLaunchInputs, hostwrite.Work) error
	EmptyAuthority(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
	Publish(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
}

// ClaimPublication is what the assembly hands a fenced generation publisher
// (claim §6.2a, moves §7.14 marker row; claim Plan 9 D5).
type ClaimPublication struct {
	// Sync is the session's current replicator's SyncAndWait: it ships what
	// the app has committed and waits for the remote position. The publisher
	// runs it before selecting the generation's root.
	Sync func(context.Context) error
	// Check is the publisher's fence: its producer is not drained, the
	// attempt's lease is live and the captured selection still names the
	// admitted session. It runs immediately before every marker request.
	Check func(context.Context) error
}

// ClaimPublicationReceipt identifies an acknowledged marker write: the
// version it created, or found on replay, and the exact root it named.
type ClaimPublicationReceipt struct {
	MarkerVersion string
	Root          genselect.File
}

// ErrClaimRootPending is a zero-position empty bootstrap before the app's
// first transaction: the generation has no root, so no marker may name it.
var ErrClaimRootPending = errors.New("host claims: the generation has no durable root yet")

// ClaimFencedPublication writes generation.json for an admitted fresh
// generation under a ClaimPublication and returns its receipt.
type ClaimFencedPublication interface {
	PublishFenced(context.Context, hostclaim.Observed, ClaimLaunchInputs, ClaimPublication) (ClaimPublicationReceipt, error)
}

// claimSyncer is the part of lease.Replicator the publisher uses.
type claimSyncer interface {
	SyncAndWait(context.Context) error
}

// ClaimWakeAssembly owns cold restore, lease transfer and failure cleanup.
// Root supplies Stop and Launch; moves supplies only its input/publication API.
type ClaimWakeAssembly struct {
	seedFactory              filesSeedRuntimeFactory
	ServingPrerequisites     func(context.Context, registry.Route, hostclaim.Observed) error
	recoveryIntent           func(context.Context, hostclaim.Observed, hostclaim.Admission) (hostclaim.Admission, error)
	recoveryPrior            func(context.Context, hostclaim.Observed) (hostclaim.Observed, bool, error)
	recoveryPrelaunched      func(context.Context, hostclaim.Observed) error
	retained                 []*retainedClaimWork
	AfterAdmission           func(context.Context, hostclaim.Observed, hostclaim.Observed) error
	BeforeAdmission          func(context.Context, hostclaim.Observed) error
	Roles                    *ClaimRoleAssembly
	LeaseRegion              string
	Producers                func(context.Context, hostclaim.Observed) (*ClaimProducers, error)
	Inputs                   ClaimWakeInputs
	Manager                  *cgroup.Manager
	Retries                  activator.ServingRetries
	Stop                     func(context.Context, hostclaim.Observed) error
	RetireArtifacts          func(context.Context, hostclaim.Observed, ClaimLaunchInputs, string) error
	RecoverArtifacts         func(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
	RequireServingEnrollment bool
	BeforeLegacyInstall      func(context.Context, hostclaim.Observed, ClaimLaunchInputs, hostwrite.Work) error
	BeforeInstallLocked      func(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
	// Prelaunch journals the admitted session's recoverable record before
	// restore or materialization; Launch later completes that same record.
	PrepareState func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs) error
	Prelaunch    func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs) (*activator.RoleLaunch, error)
	Launch       func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error)
	Ready        func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs) error
	Retain       func(hostclaim.Observed, activator.Proc, activator.Rep, error)
	ObserveLoss  func(hostclaim.Observed, error)
	// The moves C1 integration (claim Plan 9 Group D), nil unless both
	// CARLOS_HOST_CLAIM and CARLOS_MOVES_PHASE1 are on. Continue is the
	// session forward-step table (D1); nil keeps main's phase equality.
	Continue func(kind, from, to string) bool
	// Operations and ProbeReady admit and bind landing writers (D2, D3); nil
	// keeps main's refusal of every probe and operation.
	Operations OperationAttempts
	ProbeReady ClaimProbeReadiness
	// Publication is the fenced generation publisher (D5); nil keeps main's
	// unfenced Inputs.Publish. Root binds it only with both flags on.
	Publication ClaimFencedPublication
	mu          sync.Mutex
	pending     map[string]*claimWakeAttempt
}
type claimWakeAttempt struct {
	resumedAdmission bool
	releaseMu        sync.Mutex
	prelaunch        *activator.RoleLaunch
	launch           *activator.RoleLaunch
	repDone          chan struct{}
	repErr           error
	released         bool
	observed         hostclaim.Observed
	inputs           ClaimLaunchInputs
	held             claimHeldLease
	rep              *lease.Replicator
	syncer           claimSyncer
	// started is the attempt the latest operation start passed to Start.
	started *hostclaim.WakeRetryAttempt
	// publisher closes when the fenced publisher has joined; receipt and
	// pubErr are its outcome.
	publisher chan struct{}
	receipt   *ClaimPublicationReceipt
	pubErr    error
	ctx       context.Context
	cancel    context.CancelFunc
	done      chan struct{}
	mu        sync.Mutex
	// owner is the assembly that admitted the attempt (StopWriters reads its
	// producers).
	owner *ClaimWakeAssembly
	// The writers-only stop (claim Plan 11, R2). writersMu serialises
	// StopWriters; writersClosed refuses new writer work once it began;
	// writers is its witness once it succeeded. The journal and containment
	// are the session's launch record's.
	writersMu      sync.Mutex
	writersClosed  bool
	writers        *ClaimWritersDrained
	writersJournal claimWritersJournal
	writersCgroup  claimWritersCgroup
	// replicatorFDs are the replicator's descriptors as this process saw
	// them open (R3's agent half): root's narrowing hint, never evidence.
	// replicatorFDsErr is why they could not be taken; root then exempts
	// nothing and the observation fails closed.
	replicatorFDs    []stopevidence.DescriptorHint
	replicatorFDsErr error
}

func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource, writer hostbox.Writer) (*activator.ClaimRuntime, error) {
	if w != nil && w.RequireServingEnrollment && (w.BeforeLegacyInstall == nil || w.BeforeInstallLocked == nil || w.RecoverArtifacts == nil) {
		return nil, &ClaimDependencyError{Contracts: []string{"serving artifact enrollment and root retirement"}}
	}
	if w == nil || w.LeaseRegion == "" || w.AfterAdmission == nil || w.Inputs == nil || w.Manager == nil || w.Stop == nil || w.RetireArtifacts == nil || w.Retain == nil || w.ObserveLoss == nil || w.Producers == nil {
		return nil, hostwrite.ErrUnavailable
	}
	if w.Launch == nil || w.Prelaunch == nil || w.PrepareState == nil {
		if w.Roles == nil {
			return nil, hostwrite.ErrUnavailable
		}
		if w.PrepareState == nil {
			w.PrepareState = w.Roles.PrepareState
		}
		if w.Launch == nil {
			w.Launch = w.Roles.Launch
		}
		if w.Prelaunch == nil {
			w.Prelaunch = w.Roles.Prelaunch
		}
	}
	if w.Retries.ExactHealth == nil {
		w.Retries.ExactHealth = w.exactHealth
	}
	w.mu.Lock()
	if w.pending != nil {
		w.mu.Unlock()
		return nil, hostwrite.ErrUnavailable
	}
	w.pending = map[string]*claimWakeAttempt{}
	w.mu.Unlock()
	if w.Ready == nil {
		w.Ready = claimSocketReady
	}
	read := func(ctx context.Context, h string) (hostclaim.Observed, error) {
		return hostclaim.Read(ctx, writer.Claims, h)
	}
	runtime := &activator.ClaimRuntime{Box: writer.Box, Read: read, RecoveryAttemptsReady: w.recoveryAttemptsReady}
	runtime.ResumedAdmission = func(call context.Context, prior, admitted hostclaim.Observed) error {
		a := w.get(admitted.Host)
		if a == nil || !hostclaim.SameWriterIdentity(prior, admitted) {
			return hostbox.ErrChanged
		}
		current, err := read(call, admitted.Host)
		if err != nil {
			return err
		}
		a.mu.Lock()
		defer a.mu.Unlock()
		if !a.resumedAdmission || !hostclaim.SameWriterIdentity(a.observed, admitted) || !hostclaim.SameWriterIdentity(current, admitted) {
			return hostbox.ErrChanged
		}
		a.resumedAdmission = false
		return nil
	}
	runtime.StopForeign = func(ctx context.Context, h string) error {
		o, err := read(ctx, h)
		if err != nil {
			return err
		}
		return w.Stop(ctx, o)
	}
	runtime.Prerequisites = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) error {
		if err := w.Roles.Preflight(call, ClaimLaunchInputs{}); err != nil {
			return err
		}
		if o.Claim == nil {
			return hostbox.ErrChanged
		}
		if err := w.Roles.PreflightKind(call, o.Claim.Op.Kind); err != nil {
			return err
		}
		if err := w.admitsOperation(o, p, writer.Box); err != nil {
			return err
		}
		if rt.Seed != "" || rt.IsSidecar() || rt.Tunnel != "" {
			return hostbox.ErrUnsupported
		}
		prior, _, err := w.pendingRecovery(call, o)
		if err != nil {
			return err
		}
		in, err := w.Inputs.Select(call, rt, prior, p)
		if err != nil {
			return err
		}
		if (rt.Unit != "" || rt.Backing != "") && in.Unit == nil {
			return hostbox.ErrUnsupported
		}
		if len(in.Roles) != 1 || in.Roles[0] != "main" || in.Executable == "" || len(in.ExecutableSHA256) != 64 || in.DBPath == "" || in.Identity.Config == "" || in.Identity.Release == "" || in.Admission.Mode != "cold" || (in.Baseline == nil && !in.Empty) {
			return hostbox.ErrUnsupported
		}
		if o.Claim == nil || o.Claim.Holder == nil {
			return hostbox.ErrChanged
		}
		if err := w.Roles.PreflightLaunch(rt.Host, in); err != nil {
			return err
		}
		// The predecessor must be stopped before another lease or session opens.
		if err := w.stopPredecessor(call, o); err != nil {
			return err
		}
		bounded, cancel := context.WithTimeout(call, hostbox.RequestLimit)
		held, err := lease.Acquire(bounded, writer.Bucket, o.Claim.Holder.Location, w.LeaseRegion)
		cancel()
		if err != nil {
			return err
		}
		live, stop := context.WithCancel(ctx)
		a := &claimWakeAttempt{owner: w, observed: o, inputs: in, held: held, ctx: live, cancel: stop, done: make(chan struct{})}
		w.mu.Lock()
		previous := w.pending[rt.Host]
		if previous == nil {
			w.pending[rt.Host] = a
		}
		w.mu.Unlock()
		if previous != nil {
			stop()
			return errors.Join(hostbox.ErrChanged, held.Release(call))
		}
		go func() {
			select {
			case <-held.Lost():
				stop()
				a.mu.Lock()
				current := a.observed
				a.mu.Unlock()
				close(a.done)
				cleanup, cancel := context.WithTimeout(context.Background(), 30*time.Second)
				err := w.stopAttempt(cleanup, a, current)
				cancel()
				w.ObserveLoss(current, errors.Join(a.leaseErr(), err))
			case <-live.Done():
				close(a.done)
			}
		}()
		return nil
	}
	runtime.Admit = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (result hostclaim.Observed, resultErr error) {
		a := w.get(rt.Host)
		if a == nil {
			return hostclaim.Observed{}, hostwrite.ErrUnavailable
		}
		// ServingRetries owns serving cleanup. Operations bypass that wrapper,
		// so an admission error must release their pending lease here. No
		// Prepare or app execution can have run before Admit returns. Keep the
		// claim and all durable reservations intact, including an uncertain CAS.
		if w.operationKind(o) {
			defer func() {
				if resultErr == nil {
					return
				}
				cleanup, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
				defer cancel()
				current, _ := a.snapshot()
				stopErr := w.stopAttempt(cleanup, a, current)
				resultErr = errors.Join(resultErr, stopErr, w.release(cleanup, rt.Host, a))
			}()
		}
		if err := a.leaseErr(); err != nil {
			return hostclaim.Observed{}, err
		}
		// The retry reservation changed the ETag. Re-select the typed provider
		// against that observation; never re-use authorization from the old read.
		prior, resuming, err := w.pendingRecovery(call, o)
		if err != nil {
			return hostclaim.Observed{}, err
		}
		in, err := w.Inputs.Select(call, rt, prior, p)
		if err != nil {
			return hostclaim.Observed{}, err
		}
		admitted := o
		if !resuming {
			if w.recoveryIntent != nil {
				in.Admission, err = w.recoveryIntent(call, o, in.Admission)
				if err != nil {
					return admitted, err
				}
			}
			admitted, err = writer.Admit(call, o, in.Admission, in.Baseline)
		} else {
			in.Admission.NewGeneration = o.Claim.Holder.Generation
		}
		if err != nil {
			return admitted, err
		}
		a.mu.Lock()
		a.observed, a.inputs = admitted, in
		a.mu.Unlock()
		if err := w.AfterAdmission(call, prior, admitted); err != nil {
			return admitted, err
		}
		if err := source.Guard.Activate(call, rt.Host, writer.Box, nil); err != nil {
			return admitted, err
		}
		// The launch record precedes the retry bind, restore and materialization.
		// Any later failure or crash leaves the exact tuple root stops and settles.
		if err := w.persistPrelaunch(call, a, source, admitted, in); err != nil {
			return admitted, err
		}
		if w.recoveryPrelaunched != nil {
			if err := w.recoveryPrelaunched(call, admitted); err != nil {
				return admitted, err
			}
		}
		a.mu.Lock()
		a.resumedAdmission = resuming
		a.mu.Unlock()
		return admitted, nil
	}
	runtime.Prepare = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (activator.ClaimStart, error) {
		a := w.get(rt.Host)
		if a == nil {
			return activator.ClaimStart{}, hostbox.ErrChanged
		}
		if err := a.admitsWriters(); err != nil {
			return activator.ClaimStart{}, err
		}
		if !a.advance(o, w.Continue) {
			return activator.ClaimStart{}, hostbox.ErrChanged
		}
		a.mu.Lock()
		pre := a.prelaunch
		a.mu.Unlock()
		// No restore or materialization runs without the pre-launch record.
		if pre == nil || pre.Target.Host != o.Host || pre.Target.Incarnation != o.Claim.Holder.Incarnation || pre.Target.Stamp != o.Claim.Holder.Stamp || pre.Target.Session != o.Claim.Holder.Session {
			return activator.ClaimStart{}, hostbox.ErrChanged
		}
		selected, err := source.SelectAt(call, rt.Host, claimSpawnAction(o.Claim), hostguard.Captured{Incarnation: o.Claim.Holder.Incarnation, Stamp: o.Claim.Holder.Stamp})
		if err != nil {
			return activator.ClaimStart{}, err
		}
		_, in := a.snapshot()
		observation := &claimRestoreObservation{observed: o, box: writer.Box, dataDir: source.Guard.DataDir, db: in.DBPath}
		in.claimWrite = observation.retag
		selected = claimLeaseWork{Work: selected, Alive: a.ctx, Check: a.leaseErr}
		restore := hostbox.ColdRestore{Reader: writer.Data, Bucket: writer.Bucket, Generation: o.Claim.Holder.Generation, Work: selected, Identity: in.Identity, RetireArtifacts: func(ctx context.Context, g string) error { return w.RetireArtifacts(ctx, o, in, g) }}
		if o.Claim.Op.Kind == hostclaim.KindServing {
			if w.BeforeLegacyInstall != nil {
				restore.BeforeLegacyInstall = func(ctx context.Context, path, generation string) error {
					if path != in.DBPath || generation != o.Claim.Holder.Generation {
						return hostbox.ErrChanged
					}
					return w.BeforeLegacyInstall(ctx, o, in, selected)
				}
			}
			if w.BeforeInstallLocked != nil {
				restore.BeforeInstallLocked = func(ctx context.Context, path, generation string) error {
					if path != in.DBPath || generation != o.Claim.Holder.Generation {
						return hostbox.ErrChanged
					}
					return w.BeforeInstallLocked(ctx, o, in)
				}
			}
		}
		s := activator.ClaimStart{Live: a.ctx, Work: selected, Identity: in.Identity, Manager: w.Manager, Host: rt.Host, Limit: in.Limit, Version: in.Version, Staged: in.Staged}
		s.Restore = func(call context.Context) error {
			if err := a.admitsWriters(); err != nil {
				return err
			}
			if w.RecoverArtifacts != nil {
				if err := w.RecoverArtifacts(call, o, in); err != nil {
					return err
				}
			}
			if err := w.PrepareState(call, o, selected, in); err != nil {
				return err
			}
			if err := a.leaseErr(); err != nil {
				return err
			}
			if in.Empty {
				return restore.InstallEmpty(call, in.DBPath, func(ctx context.Context) error { return w.Inputs.EmptyAuthority(ctx, o, in) })
			}
			return restore.Install(call, in.DBPath, in.Baseline)
		}
		s.RetryClaimWrite = observation.retry
		s.Materialize = func(ctx context.Context) error {
			expected := observation.snapshot()
			fresh, err := read(ctx, o.Host)
			if err != nil {
				return err
			}
			if !sameClaimObservation(expected, fresh) {
				return hostclaim.ErrConflict
			}
			if !a.advance(expected, w.Continue) {
				return hostbox.ErrChanged
			}
			return w.Inputs.Materialize(ctx, expected, in, selected)
		}

		s.Build = func() *exec.Cmd {
			cmd := exec.Command(in.Executable, in.Args...)
			cmd.Env = append([]string(nil), in.Env...)
			return cmd
		}
		ready, err := w.readiness(o, p, rt, in, a, pre)
		if err != nil {
			return activator.ClaimStart{}, err
		}
		s.Ready = ready
		s.PrepareLaunch = func(ctx context.Context) (*activator.RoleLaunch, error) {
			if err := a.admitsWriters(); err != nil {
				return nil, err
			}
			launch, err := w.Launch(ctx, o, selected, in, pre)
			if err != nil {
				return nil, err
			}
			if launch == nil {
				return nil, hostwrite.ErrUnavailable
			}
			producers, err := w.Producers(ctx, o)
			if err != nil {
				return nil, err
			}
			launch.RegisterProducer = func(call context.Context, id string) (activator.RoleProducer, error) {
				if err := call.Err(); err != nil {
					return activator.RoleProducer{}, err
				}
				producerCtx, finish, err := producers.Register(a.ctx, launch.Target, id)
				return activator.RoleProducer{Context: producerCtx, Finish: finish}, err
			}
			a.mu.Lock()
			a.launch = launch
			a.mu.Unlock()
			a.bindWriters(launch)
			return launch, nil
		}
		s.Replicate = func(call context.Context, prefix string) (activator.Rep, error) {
			// Prepared generation markers precede replication; ordinary markers
			// follow the first durable root through s.Publish below.
			if in.Prepared {
				if err := w.Inputs.Publish(call, o, in); err != nil {
					return nil, err
				}
			}
			check := func(ctx context.Context) error { return selected.With(ctx, in.Identity, func() error { return nil }) }
			if err := a.held.BindGeneration(prefix, check); err != nil {
				return nil, err
			}
			producers, err := w.Producers(call, o)
			if err != nil {
				return nil, err
			}
			producerCtx, finished, err := producers.Register(a.ctx, a.launch.Target, ClaimProducerReplicator)
			if err != nil {
				return nil, err
			}
			var rep *lease.Replicator
			hints, hintErr := recordReplicatorHints(call, "/proc", os.Getpid(), in.DBPath, func() error {
				var err error
				rep, err = a.held.Replicate(producerCtx, in.DBPath, 0)
				return err
			})
			if rep == nil {
				finished(hintErr)
				return nil, hintErr
			}
			a.mu.Lock()
			a.replicatorFDs, a.replicatorFDsErr = hints, hintErr
			a.rep, a.syncer = rep, rep
			a.repDone = make(chan struct{})
			a.mu.Unlock()
			go func() {
				<-producerCtx.Done()
				cleanup, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
				err := rep.Close(cleanup)
				cancel()
				a.mu.Lock()
				a.repErr = err
				a.mu.Unlock()
				finished(err)
				close(a.repDone)
			}()
			return &claimOwnedRep{attempt: a, owner: w, host: rt.Host, rep: rep}, nil
		}
		s.Publish = func(ctx context.Context, current hostclaim.Observed) (hostclaim.Observed, error) {
			if !in.Prepared && w.Publication != nil {
				return w.publish(ctx, a, o, current, in, selected, read, writer.Published)
			}
			if !in.Prepared {
				if err := w.Inputs.Publish(ctx, current, in); err != nil {
					return hostclaim.Observed{}, err
				}
			}
			return writer.Published(ctx, current)
		}
		s.Retain = func(p activator.Proc, r activator.Rep, err error) { w.Retain(o, p, r, err) }
		if o.Claim.Op.Kind != hostclaim.KindServing {
			if err := w.bindOperation(&s, OperationStart{Admitted: o, Policy: p, Inputs: in, Owner: pre.Target}); err != nil {
				return activator.ClaimStart{}, err
			}
			// Probe readiness finds the role cgroup by the exact attempt this
			// start passed to Start, never by another CurrentAttempt read.
			current := s.CurrentAttempt
			s.CurrentAttempt = func() hostclaim.WakeRetryAttempt {
				attempt := current()
				a.mu.Lock()
				a.started = &attempt
				a.mu.Unlock()
				return attempt
			}
		}
		return s, nil
	}
	runtime.FailedAdmission = func(ctx context.Context, o hostclaim.Observed, cause error) error {
		a := w.get(o.Host)
		if a == nil {
			return nil
		}
		current, _ := a.snapshot()
		if err := w.stopAttempt(ctx, a, current); err != nil {
			return err
		}
		return w.release(ctx, o.Host, a)
	}
	wrapped, err := w.Retries.Wrap(runtime)
	if err != nil {
		return nil, err
	}
	if w.ServingPrerequisites != nil {
		previous := wrapped.Prerequisites
		wrapped.Prerequisites = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) error {
			if !p.Probe {
				if err := w.ServingPrerequisites(ctx, rt, o); err != nil {
					return err
				}
			}
			return previous(ctx, rt, o, p)
		}
	}
	return wrapped, nil
}

// claimHeldLease is the part of lease.Held one admitted attempt uses.
type claimHeldLease interface {
	Err() error
	Lost() <-chan struct{}
	Release(context.Context) error
	BindGeneration(string, func(context.Context) error) error
	Replicate(context.Context, string, time.Duration) (*lease.Replicator, error)
}

// persistPrelaunch records the admitted session under the captured guard. It
// runs once per attempt and binds exactly the admitted writer session.
func (w *ClaimWakeAssembly) persistPrelaunch(ctx context.Context, a *claimWakeAttempt, source hostbox.WorkSource, admitted hostclaim.Observed, in ClaimLaunchInputs) error {
	if w.Prelaunch == nil || admitted.Claim == nil || admitted.Claim.Holder == nil {
		return hostwrite.ErrUnavailable
	}
	a.mu.Lock()
	existing := a.prelaunch
	a.mu.Unlock()
	if existing != nil {
		return hostbox.ErrChanged
	}
	h := admitted.Claim.Holder
	selected, err := source.SelectAt(ctx, admitted.Host, claimSpawnAction(admitted.Claim), hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp})
	if err != nil {
		return err
	}
	pre, err := w.Prelaunch(ctx, admitted, claimLeaseWork{Work: selected, Alive: a.ctx, Check: a.leaseErr}, in)
	if err != nil {
		return err
	}
	if pre == nil || pre.Journal == nil || pre.Target.Host != admitted.Host || pre.Target.Incarnation != h.Incarnation || pre.Target.Stamp != h.Stamp || pre.Target.Session != h.Session {
		if pre != nil && pre.Parent != nil {
			return errors.Join(hostbox.ErrChanged, pre.Parent.Close())
		}
		return hostbox.ErrChanged
	}
	a.mu.Lock()
	a.prelaunch = pre
	a.mu.Unlock()
	a.bindWriters(pre)
	return nil
}

// bindWriters records the launch record's journal and containment for
// StopWriters. A record without either leaves the binding unchanged.
func (a *claimWakeAttempt) bindWriters(l *activator.RoleLaunch) {
	if l == nil || l.Journal == nil || l.Parent == nil {
		return
	}
	a.mu.Lock()
	a.writersJournal, a.writersCgroup = l.Journal, l.Parent
	a.mu.Unlock()
}

// publish runs a fresh generation's marker publication as a producer of the
// session's exact launch (claim §6.2a): root's fence cancels it and joins it
// before acknowledging the drain, and its check runs before every marker
// request. The publisher syncs the current replicator before it selects the
// root. A zero-position empty bootstrap has no root until the app's first
// transaction (R8-3, moves §7.14 "the writer waits"): the same producer
// keeps waiting for it in the background, and the claim records publication
// only from its receipt.
func (w *ClaimWakeAssembly) publish(ctx context.Context, a *claimWakeAttempt, admitted, current hostclaim.Observed, in ClaimLaunchInputs, selected hostwrite.Work, read func(context.Context, string) (hostclaim.Observed, error), published func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)) (hostclaim.Observed, error) {
	a.mu.Lock()
	launch, syncer, previous := a.launch, a.syncer, a.publisher
	a.mu.Unlock()
	if w.Publication == nil || launch == nil || syncer == nil || selected == nil || read == nil || published == nil {
		return hostclaim.Observed{}, hostwrite.ErrUnavailable
	}
	if previous != nil || !a.advance(current, w.Continue) {
		return hostclaim.Observed{}, hostbox.ErrChanged
	}
	producers, err := w.Producers(ctx, admitted)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	producerCtx, finish, err := producers.Register(a.ctx, launch.Target, ClaimProducerPublisher)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	done := make(chan struct{})
	a.mu.Lock()
	a.publisher = done
	a.mu.Unlock()
	// Every request has returned when the publisher settles, so it is
	// joined; its outcome is the receipt, not drain evidence.
	settle := func(receipt ClaimPublicationReceipt, err error) {
		a.mu.Lock()
		if err == nil {
			a.receipt = &receipt
		}
		a.pubErr = err
		a.mu.Unlock()
		close(done)
		finish(nil)
	}
	pub := ClaimPublication{Sync: syncer.SyncAndWait, Check: func(ctx context.Context) error {
		if err := producerCtx.Err(); err != nil {
			return err
		}
		return selected.With(ctx, in.Identity, producerCtx.Err)
	}}
	joined, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
	stop := context.AfterFunc(producerCtx, cancel)
	receipt, err := w.Publication.PublishFenced(joined, current, in, pub)
	stop()
	cancel()
	if in.Empty && errors.Is(err, ErrClaimRootPending) {
		go func() {
			settle(w.awaitRoot(producerCtx, a, admitted, in, pub, read, published))
		}()
		return current, nil
	}
	settle(receipt, err)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	return published(ctx, current)
}

// awaitRoot is the zero-position publisher's wait for the first durable
// root, with a doubling backoff from one second to thirty while the producer
// lives. Each try reads the claim afresh and is bounded like the first. The
// fence, a changed writer or a refused marker ends it without a receipt.
func (w *ClaimWakeAssembly) awaitRoot(ctx context.Context, a *claimWakeAttempt, admitted hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication, read func(context.Context, string) (hostclaim.Observed, error), published func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)) (ClaimPublicationReceipt, error) {
	delay := time.Second
	for {
		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ClaimPublicationReceipt{}, ctx.Err()
		case <-timer.C:
		}
		delay = min(2*delay, 30*time.Second)
		current, err := read(ctx, admitted.Host)
		if err == nil && !a.advance(current, w.Continue) {
			return ClaimPublicationReceipt{}, hostbox.ErrChanged
		}
		if err != nil {
			continue
		}
		bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
		receipt, err := w.Publication.PublishFenced(bounded, current, in, pub)
		cancel()
		switch {
		case errors.Is(err, ErrClaimRootPending):
			continue
		case errors.Is(err, hostbox.ErrChanged) || ctx.Err() != nil:
			return ClaimPublicationReceipt{}, errors.Join(err, ctx.Err())
		case err != nil:
			continue
		}
		// A failed claim CAS retries; the marker replay finds this receipt.
		if _, err := published(ctx, current); err == nil {
			return receipt, nil
		}
	}
}

func (w *ClaimWakeAssembly) get(host string) *claimWakeAttempt {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.pending[host]
}
func (w *ClaimWakeAssembly) release(ctx context.Context, host string, a *claimWakeAttempt) error {
	a.releaseMu.Lock()
	defer a.releaseMu.Unlock()
	a.cancel()
	select {
	case <-a.done:
	case <-ctx.Done():
		return ctx.Err()
	}
	a.mu.Lock()
	publisher := a.publisher
	a.mu.Unlock()
	// The publisher's producer context is the attempt's: it is cancelled
	// above, and its last request returns before the lease is released.
	if publisher != nil {
		select {
		case <-publisher:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	a.mu.Lock()
	done := a.repDone
	released := a.released
	previous := a.repErr
	a.mu.Unlock()
	if released {
		return previous
	}
	var err error
	if done != nil {
		select {
		case <-done:
		case <-ctx.Done():
			return ctx.Err()
		}
		a.mu.Lock()
		err = a.repErr
		a.mu.Unlock()
	} else {
		err = a.held.Release(ctx)
	}
	if err == nil {
		a.mu.Lock()
		parent := (*cgroup.Prepared)(nil)
		if a.launch != nil {
			parent = a.launch.Parent
		} else if a.prelaunch != nil {
			parent = a.prelaunch.Parent
		}
		a.mu.Unlock()
		if parent != nil {
			err = parent.Close()
		}
	}
	a.mu.Lock()
	a.released = true
	a.repErr = err
	a.mu.Unlock()
	if err == nil {
		w.mu.Lock()
		if w.pending[host] == a {
			delete(w.pending, host)
		}
		w.mu.Unlock()
	}
	return err
}
func (w *ClaimWakeAssembly) Close(ctx context.Context) error {
	w.mu.Lock()
	pending := make(map[string]*claimWakeAttempt, len(w.pending))
	for h, a := range w.pending {
		pending[h] = a
	}
	w.mu.Unlock()
	var errs []error
	for h, a := range pending {
		current, _ := a.snapshot()
		if err := w.stopAttempt(ctx, a, current); err != nil {
			errs = append(errs, err)
			continue
		}
		errs = append(errs, w.release(ctx, h, a))
	}
	return errors.Join(errs...)
}

type claimOwnedRep struct {
	attempt *claimWakeAttempt
	owner   *ClaimWakeAssembly
	host    string
	rep     *lease.Replicator
}

func (r *claimOwnedRep) Err() error { return r.rep.Err() }

// Close is the activator's stop of the replica. Once StopWriters has begun,
// the replicator belongs to the writers-stopped release: root's stop drains
// it with the other producers and Wake.Stop then releases the lease, so the
// activator's close leaves it running (claim Plan 11, R2).
func (r *claimOwnedRep) Close(ctx context.Context) error {
	r.attempt.mu.Lock()
	stopped := r.attempt.writersClosed
	r.attempt.mu.Unlock()
	if stopped {
		return nil
	}
	return r.owner.release(ctx, r.host, r.attempt)
}

// claimWakeContinue is the wake and selected-work checkpoint rule: an admitted
// writer may observe a later phase of its own row after missing one (m-2). The
// artifact retirement and the claim's own transitions keep SessionContinues.
var claimWakeContinue = hostclaim.SessionObserves

// claimSameWriter reports whether b, observed at or after a, still names a's
// admitted writer session. continues is the session forward-step table; nil
// requires the same phase, as main did.
func claimSameWriter(a, b hostclaim.Observed, continues func(kind, from, to string) bool) bool {
	if a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil {
		return false
	}
	x, y := a.Claim, b.Claim
	phase := x.Op.Phase == y.Op.Phase
	if continues != nil {
		phase = continues(x.Op.Kind, x.Op.Phase, y.Op.Phase)
	}
	return a.Host == b.Host && 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 && phase && x.Op.Kind == y.Op.Kind && x.Op.Exec == y.Op.Exec
}

// admitsOperation is the probe and operation admission path (claim Plan 9
// D2, claim Plan 11 R1). Ordinary serving wakes keep main's path unchanged.
// With the moves C1 owners bound, the K6 table admits a landing probe while
// the op is at probing and a landing wake once it is committed; with region
// HA's relocation owners also bound, it admits a relocation target's probe
// at probing where op.exec is this box, and no ordinary relocating wake.
// Every other probe or operation keeps main's typed refusal, before any
// lease, stop or reservation.
func (w *ClaimWakeAssembly) admitsOperation(o hostclaim.Observed, p activator.StartPolicy, box string) error {
	if !p.Probe && (o.Claim == nil || o.Claim.Op.Kind == hostclaim.KindServing) {
		return nil
	}
	ops, ready := w.operationOwner(o)
	if ops == nil || (p.Probe && ready == nil) || (!p.Probe && o.Claim != nil && claimRecoveryKind(o.Claim.Op.Kind)) {
		return claimOperationRefusal()
	}
	return hostbox.WakePhase(o.Claim, box, p.Probe)
}

// bindOperation binds the moves attempt journal to one admitted operation
// writer, as ServingRetries does for serving. The closures carry the exact
// admitted observation; moves refuses once it no longer holds.
func (w *ClaimWakeAssembly) bindOperation(s *activator.ClaimStart, start OperationStart) error {
	ops, _ := w.operationOwner(start.Admitted)
	if ops == nil || s == nil || s.Ready == nil {
		return claimOperationRefusal()
	}
	ready := s.Ready
	if start.Policy.Probe {
		s.OpenProbe = func(ctx context.Context, proposed time.Time) (time.Time, error) {
			return ops.OpenProbe(ctx, start, proposed)
		}
	}
	s.CurrentAttempt = func() hostclaim.WakeRetryAttempt { return ops.CurrentAttempt(start) }
	s.Reserve = func(ctx context.Context, n int) error { return ops.Reserve(ctx, start, n) }
	s.Failed = func(ctx context.Context, n int) error { return ops.Failed(ctx, start, n) }
	s.Healthy = func(ctx context.Context, n int, cmd *exec.Cmd) error {
		return ops.Healthy(ctx, start, n, cmd, func(ctx context.Context) error { return ready(ctx, cmd) })
	}
	return nil
}

// readiness selects the readiness check by op kind (claim Plan 9 D3).
// Serving and a landing wake after the commit keep main's check: GET /, any
// status below 500, the exact spawned pid. A landing probe uses the probe
// check against the started main role's cgroup, which the role's journalled
// reservation names: the exact attempt the start passed to Start.
func (w *ClaimWakeAssembly) readiness(o hostclaim.Observed, p activator.StartPolicy, rt registry.Route, in ClaimLaunchInputs, a *claimWakeAttempt, pre *activator.RoleLaunch) (func(context.Context, *exec.Cmd) error, error) {
	if !p.Probe {
		return func(ctx context.Context, cmd *exec.Cmd) error {
			a.mu.Lock()
			launch := a.launch
			a.mu.Unlock()
			if launch != nil && launch.Root != nil {
				if in.Unit != nil {
					return nil
				}
				return claimPollReady(ctx, rt.Socket, "http://carlos/", func(conn net.Conn) error { return launch.PeerInRole(ctx, conn, cmd) }, func(status int) bool { return status < 500 })
			}
			return w.Ready(ctx, cmd, rt, in)
		}, nil
	}
	ops, probeReady := w.operationOwner(o)
	if ops == nil || probeReady == nil || pre == nil {
		return nil, claimOperationRefusal()
	}
	return func(ctx context.Context, cmd *exec.Cmd) error {
		a.mu.Lock()
		launch, started := a.launch, a.started
		a.mu.Unlock()
		if launch == nil || started == nil {
			return hostwrite.ErrUnavailable
		}
		group, err := launch.RoleCgroup(ctx, cmd, *started)
		if err != nil {
			return err
		}
		return probeReady(ctx, cmd, rt, in, group)
	}, nil
}

func (a *claimWakeAttempt) snapshot() (hostclaim.Observed, ClaimLaunchInputs) {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.observed, a.inputs
}

// advance is the attempt's session checkpoint (M9): next replaces the last
// checked observation only if it still names the same writer one adjacent
// forward step on at most (claimSameWriter under continues). A skipped or
// backward step refuses and leaves the checkpoint as it was.
func (a *claimWakeAttempt) advance(next hostclaim.Observed, continues func(kind, from, to string) bool) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	if !claimSameWriter(a.observed, next, continues) {
		return false
	}
	a.observed = next
	return true
}
func (a *claimWakeAttempt) leaseErr() error {
	if err := a.ctx.Err(); err != nil {
		return err
	}
	a.mu.Lock()
	rep := a.rep
	a.mu.Unlock()
	if rep != nil {
		return rep.Err()
	}
	return a.held.Err()
}

type claimLeaseWork struct {
	Work  hostwrite.Work
	Alive context.Context
	Check func() error
}

func (w claimLeaseWork) With(ctx context.Context, id hostwrite.Identity, fn func() error) error {
	if w.Work == nil || w.Alive == nil || w.Check == nil || fn == nil {
		return hostwrite.ErrUnavailable
	}
	joined, cancel := context.WithCancel(ctx)
	defer cancel()
	stop := context.AfterFunc(w.Alive, cancel)
	defer stop()
	return w.Work.With(joined, id, func() error {
		if err := w.Alive.Err(); err != nil {
			return err
		}
		if err := joined.Err(); err != nil {
			return err
		}
		if err := w.Check(); err != nil {
			return err
		}
		return fn()
	})
}
func claimSocketReady(ctx context.Context, cmd *exec.Cmd, rt registry.Route, _ ClaimLaunchInputs) error {
	if cmd == nil || cmd.Process == nil || rt.Socket == "" {
		return hostwrite.ErrUnavailable
	}
	pid := cmd.Process.Pid
	return claimPollReady(ctx, rt.Socket, "http://carlos/", func(conn net.Conn) error { return claimReadyPeer(conn, pid) }, func(status int) bool { return status < 500 })
}

// claimPollReady polls target over the role's socket until accept takes the
// response status. Every connection's peer is verified before a request is
// written on it; redirects are answers, never followed.
func claimPollReady(ctx context.Context, socket, target string, peer func(net.Conn) error, accept func(int) bool) error {
	transport := &http.Transport{Proxy: nil, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
		conn, err := (&net.Dialer{}).DialContext(ctx, "unix", socket)
		if err != nil {
			return nil, err
		}
		if err = peer(conn); err != nil {
			conn.Close()
			return nil, err
		}
		return conn, nil
	}, DisableKeepAlives: true}
	defer transport.CloseIdleConnections()
	client := &http.Client{Transport: transport, Timeout: time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
	timer := time.NewTicker(150 * time.Millisecond)
	defer timer.Stop()
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
		if err != nil {
			return err
		}
		resp, err := client.Do(req)
		if err == nil {
			resp.Body.Close()
			if accept(resp.StatusCode) {
				return nil
			}
			err = fmt.Errorf("readiness HTTP %d", resp.StatusCode)
		}
		select {
		case <-ctx.Done():
			return errors.Join(ctx.Err(), err)
		case <-timer.C:
		}
	}
}

func (w *ClaimWakeAssembly) exactHealth(ctx context.Context, cmd *exec.Cmd, attempt hostclaim.WakeRetryAttempt) error {
	w.mu.Lock()
	pending := make([]*claimWakeAttempt, 0, len(w.pending))
	for _, a := range w.pending {
		pending = append(pending, a)
	}
	w.mu.Unlock()
	for _, a := range pending {
		a.mu.Lock()
		launch := a.launch
		a.mu.Unlock()
		if launch != nil {
			if err := launch.ExactHealth(ctx, cmd, attempt); err == nil {
				return nil
			}
		}
	}
	return hostbox.ErrChanged
}

// stopPredecessor uses sealed cross-box evidence only when production wired
// that path. The ordinary local stop remains the default.
func (w *ClaimWakeAssembly) pendingRecovery(ctx context.Context, o hostclaim.Observed) (hostclaim.Observed, bool, error) {
	if w.recoveryPrior == nil {
		return o, false, nil
	}
	return w.recoveryPrior(ctx, o)
}
func (w *ClaimWakeAssembly) stopPredecessor(ctx context.Context, o hostclaim.Observed) error {
	prior, pending, err := w.pendingRecovery(ctx, o)
	if err != nil {
		return err
	}
	if pending {
		return w.AfterAdmission(ctx, prior, o)
	}
	if w.BeforeAdmission != nil {
		return w.BeforeAdmission(ctx, o)
	}
	return w.Stop(ctx, o)
}

// stopAttempt never sends a predecessor on another box to the local root.
// An operation that has not made a prelaunch record cannot have launched any
// roles or started replication. Releasing that unused lease is not a stop
// receipt or a refund: the claim, effect and attempt journals remain intact.
func (w *ClaimWakeAssembly) stopAttempt(ctx context.Context, a *claimWakeAttempt, current hostclaim.Observed) error {
	a.mu.Lock()
	pre := a.prelaunch
	a.mu.Unlock()
	if w.Continue != nil && pre == nil && w.operationKind(current) {
		return nil
	}
	// Session zero admitted no writer; the predecessor witness already ran.
	if pre == nil && current.Claim != nil && current.Claim.Holder != nil && current.Claim.Holder.Session == 0 {
		return nil
	}
	return w.Stop(ctx, current)
}

// claimRecoveryKind identifies the source recovery operations.
func claimRecoveryKind(kind string) bool {
	return kind == hostclaim.KindReverting || kind == hostclaim.KindRestoring
}