package host

import (
	"context"
	"encoding/json"
	"errors"
	"github.com/carlosframework/platform/internal/park"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"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/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// ClaimKindOwners is one family of op kinds' input, admission and
// publication owners (claim Plan 9 D5, moves ruling H16).
type ClaimKindOwners struct {
	InputPolicy         func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error
	Inputs              ClaimWakeInputs
	AdmissionEvidence   func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
	PublicationEvidence func(context.Context, hostclaim.Observed) error
}

func (o ClaimKindOwners) complete() bool {
	return o.Inputs != nil && o.AdmissionEvidence != nil && o.PublicationEvidence != nil
}

// ClaimMovesOwners is the moves integration's owner set. The dispatcher hands
// it only moving, landing, reverting and restoring observations, and only
// with CARLOS_MOVES_PHASE1 on. ProbeReady replaces the claim's probe
// readiness check when set.
type ClaimMovesOwners struct {
	LossRecovery hostclaim.LossRecoveryOwner
	LossSources  hostbox.LossAdmissionSources
	ClaimKindOwners
	Operations OperationAttempts
	ProbeReady ClaimProbeReadiness
}

func (m ClaimMovesOwners) zero() bool {
	return m.LossRecovery == nil && m.LossSources == nil && m.Inputs == nil && m.AdmissionEvidence == nil && m.PublicationEvidence == nil && m.Operations == nil && m.ProbeReady == nil
}

// ClaimRelocationOwners is region HA's owner set for the relocating kind
// (claim Plan 11 R1, R8). Inputs is region HA's host.RelocationWakeInputs;
// with the admission and publication evidence it is one ClaimKindOwners
// family, exactly as the moves owners are. Operations is the relocation's
// attempt journal for relocating/probing (D2's OperationAttempts contract).
// ProbeReady replaces the claim's probe readiness check when set. Policy is
// the typed hostbox.Policy.Relocation owner: it runs only after the claim's
// own relocating gate passed, so it can only refuse. The dispatcher hands
// these owners only relocating observations at a phase of the relocating
// session row, and only with CARLOS_HOST_CLAIM, CARLOS_MOVES_PHASE1 and
// CARLOS_REGION_HA all on.
type ClaimRelocationOwners struct {
	// These hooks verify relocation's source release and target installation.
	BeforeAdmission func(context.Context, hostclaim.Observed) error
	AfterAdmission  func(context.Context, hostclaim.Observed, hostclaim.Observed) error
	ClaimKindOwners
	Operations OperationAttempts
	ProbeReady ClaimProbeReadiness
	Policy     func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error
}

func (r ClaimRelocationOwners) zero() bool {
	return r.BeforeAdmission == nil && r.AfterAdmission == nil && r.Inputs == nil && r.AdmissionEvidence == nil && r.PublicationEvidence == nil && r.Operations == nil && r.ProbeReady == nil && r.Policy == nil
}

// claimRelocationRefusal is the typed refusal for a relocating observation
// whose owner set is bound but incomplete.
func claimRelocationRefusal() error {
	return &ClaimDependencyError{Contracts: []string{"region HA relocation wake inputs, attempt journal and policy"}}
}

// claimRelocationC1 decides whether region HA's relocation owners are bound.
// They need the moves C1 integration (Group D: the session table, the fenced
// publisher and the operation path) and CARLOS_REGION_HA on. Either off, or
// no relocation owners, leaves relocating exactly as Plan 9 left it: refused.
// A partial owner set refuses the whole assembly rather than half-wiring it.
func claimRelocationC1(getenv func(string) string, movesC1 bool, r ClaimRelocationOwners) (bool, error) {
	if !movesC1 || getenv == nil || !regionHAEnabled(getenv) || r.zero() {
		return false, nil
	}
	if !r.complete() || r.Operations == nil || r.Policy == nil || r.BeforeAdmission == nil || r.AfterAdmission == nil {
		return false, hostwrite.ErrUnavailable
	}
	return true, nil
}

// ClaimServingLaunch is the exact launch the serving exec path would run for
// a route: the live executable and its digest, its arguments and composed
// environment, the cgroup limit, the live version, and the config and
// release identities the provider's InputPolicy accepts for that route.
type ClaimServingLaunch struct {
	Unit             *hostbox.UnitLaunchInputs
	Identity         hostwrite.Identity
	Executable       string
	ExecutableSHA256 string
	Args, Env        []string
	Version          string
	// Staged is the file ExecutableSHA256 hashes and the version it is
	// staged under, from one resolution of the live link
	// (activator.ClaimLaunchSnapshot.Staged).
	Staged release.StagedIdentity
	Limit  cgroup.Limit
	// Roles are the processes the real start spawns: "main", plus "seed"
	// and "sidecar" when the route carries them. Nil means main alone.
	Roles []string
}

// ClaimMarkerStore writes generation.json conditionally (moves §5.1, claim
// §6.4a item 5): PutMarker succeeds only while the key's current object has
// expectedETag, or, with "", while the key has no current object
// (If-None-Match). It returns the version the write created. check is the
// publisher's fence: the store runs it immediately before issuing every
// request attempt, retries included, and issues nothing once it fails. Every
// attempt runs under ctx, which bounds the request and is cancelled by the
// fence; PutMarker returns only after its last attempt has returned.
type ClaimMarkerStore interface {
	PutMarker(ctx context.Context, bucket, key string, body []byte, expectedETag string, check func(context.Context) error) (versionID string, err error)
}

// ClaimFilesManifest is the moves C8 files-manifest contract (claim §6.2a
// P6-D3, moves C8). It is external: the moves integration implements it and
// the claim has no stand-in. Establish runs once per start, after the held
// identities are proved and before the app executes, and does its local
// work only under work, the start's captured selection. It establishes the
// admitted generation's files manifest from the newest valid manifest of its
// admitted predecessor chain, read by exact version, keeping each reference
// origin-qualified to its immutable object in the incarnation's location and
// region. Missing or malformed data is an error, never an empty-files
// bootstrap; only an empty bootstrap (in.Empty, after EmptyAuthority) starts
// from no manifest.
//
// The instance key (L21, plan 2026-10-10-instance-key-provision.md
// obligation O1'): the launch source has already minted
// CARLOS_INSTANCE_KEY into the host's own <host>.env, durably, when
// Establish runs, and the app executes only after it returns. So the
// manifest Establish leaves must carry the env file (region HA §5.6's
// always-continuous platform item) with exactly the bytes on disk when it
// runs, staged and committed before it returns. A new admission starts
// with no adoption record: the manifest carries no instancekey.AdoptionItem
// (StagePlatformDelete), and the local <host>.instance-key-adoption.json is
// removed under work. An error, a bucket that cannot be reached included,
// fails the wake and the app does not start. Establish never writes the env
// file on disk.
type ClaimFilesManifest interface {
	Establish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error
}

// claimFilesRefusal is the typed refusal while no C8 provider is bound.
func claimFilesRefusal() error {
	return &ClaimDependencyError{Contracts: []string{"moves C8 origin-qualified files manifest provider"}}
}

// ClaimServingSources are the typed sources the claim's own serving owners
// read. The provider supplies them: the launch identity and InputPolicy are
// one contract, the conditional marker write is a data bucket capability,
// and Files is the required moves C8 files-manifest provider. Without Files
// every serving wake is refused, typed, before any lease or launch.
type ClaimServingSources struct {
	Launch  func(context.Context, registry.Route) (ClaimServingLaunch, error)
	Markers ClaimMarkerStore
	Files   ClaimFilesManifest
	// Identity verifies this box's own identity (the roster entry and its
	// evidence, hostbox.IdentityProvider.Verify) before every data-bucket
	// request region HA's relocation inputs make, as main's serving owners
	// do through their RequestStore's Check (claimconstruction.go). Only
	// the relocation source requires it; main binds its own.
	Identity func(context.Context) error
}

// claimMovesC1 decides whether the moves C1 integration (Group D) is bound.
// The production assembly opens only with CARLOS_HOST_CLAIM on; the
// integration also needs CARLOS_MOVES_PHASE1 on and a moves owner set.
// Moves off, or no moves owners, leaves every Group D path inert: main's
// refusals and phase equality stand. A partial owner set refuses the whole
// assembly rather than half-wiring it.
func claimMovesC1(getenv func(string) string, m ClaimMovesOwners) (bool, error) {
	if getenv == nil || !hostwrite.Enabled(getenv) || !moves.Enabled(getenv) || m.zero() {
		return false, nil
	}
	if !m.complete() || m.Operations == nil {
		return false, hostwrite.ErrUnavailable
	}
	return true, nil
}

func claimMovesKind(kind string) bool {
	switch kind {
	case hostclaim.KindMoving, hostclaim.KindLanding, hostclaim.KindReverting, hostclaim.KindRestoring:
		return true
	}
	return false
}

// ClaimOwnerDispatch routes every owner call by the observed claim's op kind:
// serving to the claim's own serving owners, the four move kinds to the
// moves owners, relocating to region HA's typed Relocation entry (claim Plan
// 11 R1). Moves nil keeps every move kind refused. Relocation nil keeps
// relocating unsupported, as before Plan 11; bound, it receives relocating
// observations only at a phase of the relocating session row (probing, and
// the passed probe's session serving), and every other relocating phase is
// refused before any owner runs. Any other kind is refused as unsupported.
type ClaimOwnerDispatch struct {
	Serving    ClaimKindOwners
	Moves      *ClaimKindOwners
	Relocation *ClaimKindOwners
}

func (d ClaimOwnerDispatch) owners(o hostclaim.Observed) (ClaimKindOwners, error) {
	if o.Absent || o.Claim == nil {
		return ClaimKindOwners{}, hostbox.ErrChanged
	}
	switch kind := o.Claim.Op.Kind; {
	case kind == hostclaim.KindServing:
		if !d.Serving.complete() {
			return ClaimKindOwners{}, hostwrite.ErrUnavailable
		}
		return d.Serving, nil
	case claimMovesKind(kind):
		if d.Moves == nil || !d.Moves.complete() {
			return ClaimKindOwners{}, claimOperationRefusal()
		}
		return *d.Moves, nil
	case kind == hostclaim.KindRelocating && d.Relocation != nil:
		if !d.Relocation.complete() {
			return ClaimKindOwners{}, claimRelocationRefusal()
		}
		if !hostclaim.SessionPhase(hostclaim.KindRelocating, o.Claim.Op.Phase) {
			return ClaimKindOwners{}, hostbox.ErrChanged
		}
		return *d.Relocation, nil
	}
	return ClaimKindOwners{}, hostbox.ErrUnsupported
}

func (d ClaimOwnerDispatch) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	owners, err := d.owners(o)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	return owners.Inputs.Select(ctx, rt, o, p)
}

func (d ClaimOwnerDispatch) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	owners, err := d.owners(o)
	if err != nil {
		return err
	}
	return owners.Inputs.Materialize(ctx, o, in, work)
}

func (d ClaimOwnerDispatch) EmptyAuthority(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	owners, err := d.owners(o)
	if err != nil {
		return err
	}
	return owners.Inputs.EmptyAuthority(ctx, o, in)
}

func (d ClaimOwnerDispatch) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	owners, err := d.owners(o)
	if err != nil {
		return err
	}
	return owners.Inputs.Publish(ctx, o, in)
}

// PublishFenced routes the fenced publisher by op kind. An owner set without
// its own fenced publisher keeps a typed refusal and never falls back to the
// unfenced Publish.
func (d ClaimOwnerDispatch) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	owners, err := d.owners(o)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	p, ok := owners.Inputs.(ClaimFencedPublication)
	if !ok {
		return ClaimPublicationReceipt{}, &ClaimDependencyError{Contracts: []string{"moves C1 fenced generation publisher"}}
	}
	return p.PublishFenced(ctx, o, in, pub)
}

func (d ClaimOwnerDispatch) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	owners, err := d.owners(o)
	if err != nil {
		return err
	}
	return owners.AdmissionEvidence(ctx, o, in, baseline)
}

func (d ClaimOwnerDispatch) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
	owners, err := d.owners(o)
	if err != nil {
		return err
	}
	return owners.PublicationEvidence(ctx, o)
}

// claimServingOwners is the claim's own serving input, admission and
// publication owner set (claim Plan 9 D5). A serving wake is a cold admission
// into a fresh generation restored from the certified remote baseline of
// holder.generation in this region (claim §6.4a; moves §7.14 serving row,
// all cold while P7 stands). The first writer of a created incarnation is the
// empty bootstrap (claim §6.4a "Empty bootstrap"). The marker names the new
// generation, with snapshot_txid 0, only after the session's own root is
// acknowledged (INV-11), and publication evidence binds the exact marker
// version this process wrote and the exact root version it named.
type claimServingOwners struct {
	emptyRecovery       claimEmptyRecovery
	HomeRegion          string
	Data                genselect.Reader
	Bucket, Region, Box string
	NewID               func() string
	Now                 func() time.Time
	Sources             ClaimServingSources
	mu                  sync.Mutex
	published           map[string]ClaimPublicationReceipt
}

func checkClaimServingSources(s ClaimServingSources) error {
	if s.Launch == nil || s.Markers == nil || s.Files == nil {
		return &ClaimDependencyError{Contracts: []string{"claimed serving launch, marker and C8 files owners"}}
	}
	return nil
}

func (s *claimServingOwners) owners() ClaimKindOwners {
	return ClaimKindOwners{Inputs: s, AdmissionEvidence: s.admissionEvidence, PublicationEvidence: s.publicationEvidence}
}

func (s *claimServingOwners) serving(o hostclaim.Observed) error {
	if s == nil || s.Data == nil || s.Bucket == "" || s.Region == "" || s.Box == "" || s.NewID == nil || s.Now == nil {
		return hostwrite.ErrUnavailable
	}
	if err := checkClaimServingSources(s.Sources); err != nil {
		return err
	}
	if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.ID != "" || o.Claim.Op.Phase != "" || o.Claim.Op.Exec != s.Box {
		return hostbox.ErrChanged
	}
	return nil
}

func (s *claimServingOwners) baseline(ctx context.Context, o hostclaim.Observed) (*genselect.Plan, error) {
	return genselect.Select(ctx, s.Data, genselect.Request{Bucket: s.Bucket, Region: s.Region, Host: o.Host, Holder: park.HolderRegionView(*o.Claim.Holder, s.HomeRegion, s.Region)})
}

// claimEmptyBootstrap: a created incarnation before its first writer.
func claimEmptyBootstrap(h *hostclaim.Holder) bool {
	return h.Session == 0 && h.Generation == "" && h.WriterBox == "" && len(h.Admissions) == 0
}

// claimFreshGeneration: a usable generation name this holder never used.
func claimFreshGeneration(h *hostclaim.Holder, g string) bool {
	if g == "" || g == "." || g == ".." || g == hostclaim.RootGeneration || len(g) > 64 || strings.ContainsAny(g, "/\\\x00\r\n") || g == h.Generation {
		return false
	}
	for _, a := range h.Admissions {
		if a.Generation == g || a.Predecessor == g {
			return false
		}
	}
	return true
}

func claimHexDigest(s string) bool {
	if len(s) != 64 {
		return false
	}
	for _, r := range s {
		if (r < '0' || r > '9') && (r < 'a' || r > 'f') {
			return false
		}
	}
	return true
}

func (s *claimServingOwners) launch(ctx context.Context, rt registry.Route) (ClaimServingLaunch, error) {
	l, err := s.Sources.Launch(ctx, rt)
	if err != nil {
		return ClaimServingLaunch{}, err
	}
	if l.Identity.Config == "" || l.Identity.Release == "" || !filepath.IsAbs(l.Executable) || !claimHexDigest(l.ExecutableSHA256) || l.Version == "" {
		return ClaimServingLaunch{}, hostwrite.ErrUnavailable
	}
	return l, nil
}

// launchDigest is the serving retry's launch-input digest, from the launch
// source alone: no baseline listing is needed to name the inputs.
func (s *claimServingOwners) launchDigest(ctx context.Context, rt registry.Route) (string, error) {
	if s == nil || s.Sources.Launch == nil {
		return "", hostwrite.ErrUnavailable
	}
	l, err := s.launch(ctx, rt)
	if err != nil {
		return "", err
	}
	return ClaimLaunchDigest(ClaimLaunchInputs{Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: l.Args, Env: l.Env})
}

func (s *claimServingOwners) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	if err := s.serving(o); err != nil {
		return ClaimLaunchInputs{}, err
	}
	// Prerequisites selects before any lease, stop or launch.
	if s.Sources.Files == nil {
		return ClaimLaunchInputs{}, claimFilesRefusal()
	}
	if p.Probe || p.Prepared != nil || p.Inputs != nil || rt.Host != o.Host || !filepath.IsAbs(rt.DBPath) {
		return ClaimLaunchInputs{}, hostbox.ErrUnsupported
	}
	l, err := s.launch(ctx, rt)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	h := o.Claim.Holder
	generation := s.NewID()
	if !claimFreshGeneration(h, generation) {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	roles := append([]string(nil), l.Roles...)
	if len(roles) == 0 {
		roles = []string{"main"}
	}
	in := ClaimLaunchInputs{Unit: l.Unit, Roles: roles, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...), DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit,
		Admission: hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: s.Box, Mode: "cold", NewGeneration: generation, Region: s.Region}}
	empty, err := s.emptyBootstrap(ctx, o)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if empty {
		if len(h.Admissions) > 0 {
			in.Admission.PredecessorRegion = h.Admissions[len(h.Admissions)-1].Region
		}
		in.Empty = true
		return in, nil
	}
	plan, err := s.baseline(ctx, o)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	in.Baseline = plan
	in.Admission.RemoteMaxTXID = plan.MaxTXID
	// The baseline's objects are in this region: genselect selects nothing
	// else. With earlier admissions AdmitWriter takes the last record's.
	in.Admission.PredecessorRegion = s.Region
	if n := len(h.Admissions); n > 0 {
		in.Admission.PredecessorRegion = h.Admissions[n-1].Region
	}
	return in, nil
}

// Materialize proves the selected config and release identities are still
// the held ones under the selected work, then has the C8 provider establish
// the admitted generation's files manifest under that same captured work.
// Config sync and release adoption materialize serving inputs under their
// own selections; a serving wake writes neither.
func (s *claimServingOwners) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	if err := s.serving(o); err != nil {
		return err
	}
	if s.Sources.Files == nil {
		return claimFilesRefusal()
	}
	if work == nil || in.Identity.Config == "" || in.Identity.Release == "" {
		return hostwrite.ErrUnavailable
	}
	if err := work.With(ctx, in.Identity, func() error { return nil }); err != nil {
		return err
	}
	return s.Sources.Files.Establish(ctx, o, in, work)
}

// EmptyAuthority is the empty bootstrap's proof, run before any local
// effect: the admitted session is the incarnation's first writer, its
// predecessor names no generation and no region, and a complete listing of
// the holder's location holds nothing. A listing error refuses.
func (s *claimServingOwners) EmptyAuthority(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if err := s.serving(o); err != nil {
		return err
	}
	h := o.Claim.Holder
	if !in.Empty || in.Baseline != nil || len(h.Admissions) == 0 || h.WriterBox != s.Box {
		return hostbox.ErrChanged
	}
	n := len(h.Admissions)
	a := h.Admissions[n-1]
	predecessor, region := "", ""
	if n > 1 {
		if s.emptyRecovery == nil {
			return hostbox.ErrChanged
		}
		yes, err := s.emptyRecovery(ctx, o, h.Admissions[:n-1])
		if err != nil {
			return err
		}
		if !yes {
			return hostbox.ErrChanged
		}
		predecessor, region = h.Admissions[n-2].Generation, h.Admissions[n-2].Region
	}
	if a.Session != h.Session || a.Box != s.Box || a.Mode != "cold" || a.Generation != h.Generation || a.Generation != in.Admission.NewGeneration || a.Predecessor != predecessor || a.PredecessorRegion != region || a.Region != s.Region {
		return hostbox.ErrChanged
	}
	return s.emptyLocation(ctx, h.Location, h.Admissions[:n-1]...)
}

func claimPublishedKey(host string, session uint64) string {
	return host + "\x00" + strconv.FormatUint(session, 10)
}

// Publish is main's unfenced publication entry. The serving marker is written
// only by PublishFenced, as a producer of the session's launch (claim §6.2a):
// an unfenced listing could take a successor's ETag and overwrite its marker.
func (s *claimServingOwners) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if err := s.serving(o); err != nil {
		return err
	}
	return hostwrite.ErrUnavailable
}

// PublishFenced writes generation.json for the admitted fresh generation
// (moves §7.14 marker row). It syncs the session's replicator first, then
// requires the generation's own root: a zero-position empty bootstrap has
// none until the app's first transaction and gets ErrClaimRootPending, never
// a zero snapshot or a fabricated root. The marker supersedes the current
// one by its exact ETag, or is created when none exists; pub.Check runs
// immediately before the write and, in the store, before every attempt. A
// replay that finds this generation's marker from this box records it.
func (s *claimServingOwners) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	if err := s.serving(o); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	return s.publishFresh(ctx, o, in, pub)
}

// publishFresh is PublishFenced after its kind check, unchanged: the one
// fresh-generation marker writer, which region HA's later probe attempts
// also use (rev 2, ruling 47).
func (s *claimServingOwners) publishFresh(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	if pub.Sync == nil || pub.Check == nil {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	h := o.Claim.Holder
	if h.Generation == "" || h.Generation != in.Admission.NewGeneration || h.WriterBox != s.Box || in.Prepared {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	if err := pub.Sync(ctx); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
	defer cancel()
	plan, err := s.baseline(bounded, o)
	if in.Empty && errors.Is(err, genselect.ErrEmpty) {
		return ClaimPublicationReceipt{}, ErrClaimRootPending
	}
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	// The generation is fresh and only this session's bound replica client
	// writes under it, so its root is the session's own acknowledged upload.
	if plan.Generation != h.Generation || plan.Root == nil || plan.Root.MinTXID != 1 || plan.Root.VersionID == "" {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	root := *plan.Root
	key := strings.TrimSuffix(h.Location, "/") + "/generation.json"
	versions, err := s.Data.ListVersions(bounded, s.Bucket, key)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	var current *claimMarkerVersion
	for _, v := range versions {
		if v.Key == key && v.IsLatest {
			if current != nil || v.DeleteMarker {
				return ClaimPublicationReceipt{}, hostbox.ErrChanged
			}
			current = &claimMarkerVersion{ID: v.VersionID, ETag: v.ETag}
		}
	}
	if current != nil && plan.Marker != nil && plan.MarkerVersion == current.ID && plan.Marker.Op == h.Generation {
		if plan.Marker.Box != s.Box || plan.Marker.Incarnation != h.Incarnation {
			return ClaimPublicationReceipt{}, hostbox.ErrChanged
		}
		receipt := ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}
		s.record(o.Host, h.Session, receipt)
		return receipt, nil
	}
	body, err := json.Marshal(genselect.Marker{Op: h.Generation, Incarnation: h.Incarnation, Host: o.Host, Box: s.Box, InstalledAt: s.Now().UTC()})
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	expected := ""
	if current != nil {
		expected = current.ETag
	}
	if err := pub.Check(bounded); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	version, err := s.Sources.Markers.PutMarker(bounded, s.Bucket, key, body, expected, pub.Check)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if version == "" {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	receipt := ClaimPublicationReceipt{MarkerVersion: version, Root: root}
	s.record(o.Host, h.Session, receipt)
	return receipt, nil
}

type claimMarkerVersion struct{ ID, ETag string }

func (s *claimServingOwners) record(host string, session uint64, receipt ClaimPublicationReceipt) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.published == nil {
		s.published = map[string]ClaimPublicationReceipt{}
	}
	s.published[claimPublishedKey(host, session)] = receipt
}

// admissionEvidence binds a serving cold admission to its observation and to
// the certified remote baseline, read again here (claim §6.4a "The baseline
// is always the certified remote").
func (s *claimServingOwners) admissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	if err := s.serving(o); err != nil {
		return err
	}
	h := o.Claim.Holder
	if in.ObservedOpID != "" || in.ObservedPhase != "" || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != s.Box || in.Region != s.Region ||
		in.Mode != "cold" || in.PredecessorCertified || in.KeepStamp || in.Coverage != "" || len(in.AdditionalWritesLost) != 0 || in.LossID != "" || !claimFreshGeneration(h, in.NewGeneration) {
		return hostbox.ErrChanged
	}
	empty, err := s.emptyBootstrap(ctx, o)
	if err != nil {
		return err
	}
	if empty {
		region := ""
		if len(h.Admissions) > 0 {
			region = h.Admissions[len(h.Admissions)-1].Region
		}
		if baseline != nil || in.RemoteMaxTXID != 0 || in.PredecessorRegion != region {
			return hostbox.ErrChanged
		}
		return nil
	}
	if baseline == nil {
		return hostbox.ErrChanged
	}
	fresh, err := s.baseline(ctx, o)
	if err != nil {
		return err
	}
	if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || in.RemoteMaxTXID != fresh.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) {
		return hostbox.ErrChanged
	}
	if len(h.Admissions) == 0 && in.PredecessorRegion != s.Region {
		return hostbox.ErrChanged
	}
	return nil
}

// publicationEvidence requires the receipt of this process's PublishFenced
// for the admitted session: the marker is exactly the version it wrote and
// names the holder's generation, incarnation, host and this box, and the
// exact root version it named still exists under the generation with its key
// current. Retention is off for a bound generation, so the session never
// removes it. A late duplicate of the same upload is another version at the
// key, and a later compaction may give genselect a better root; neither is a
// change.
func (s *claimServingOwners) publicationEvidence(ctx context.Context, o hostclaim.Observed) error {
	if err := s.serving(o); err != nil {
		return err
	}
	return s.freshEvidence(ctx, o)
}

// freshEvidence is publicationEvidence after its kind check, unchanged.
func (s *claimServingOwners) freshEvidence(ctx context.Context, o hostclaim.Observed) error {
	h := o.Claim.Holder
	s.mu.Lock()
	receipt, ok := s.published[claimPublishedKey(o.Host, h.Session)]
	s.mu.Unlock()
	prefix, err := hostbox.GenerationPrefix(*h)
	if !ok || err != nil || receipt.MarkerVersion == "" || receipt.Root.VersionID == "" || !strings.HasPrefix(receipt.Root.Key, prefix) {
		return hostbox.ErrChanged
	}
	plan, err := s.baseline(ctx, o)
	if err != nil {
		return err
	}
	if plan.Generation != h.Generation || plan.Root == nil || plan.Marker == nil || plan.MarkerVersion != receipt.MarkerVersion || plan.Marker.Op != h.Generation || plan.Marker.Box != s.Box || plan.Marker.Incarnation != h.Incarnation || plan.Marker.Host != o.Host {
		return hostbox.ErrChanged
	}
	versions, err := s.Data.ListVersions(ctx, s.Bucket, receipt.Root.Key)
	if err != nil {
		return err
	}
	exact, current := false, false
	for _, v := range versions {
		if v.Key != receipt.Root.Key || v.DeleteMarker {
			continue
		}
		exact = exact || v.VersionID == receipt.Root.VersionID
		current = current || v.IsLatest
	}
	if !exact || !current {
		return hostbox.ErrChanged
	}
	return nil
}