package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"reflect"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancekey"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/release"
)

// casBucketStore is a release.Store over one leasetest bucket: current
// reads with the latest version id as the etag, create-only and
// compare-and-swap writes. It stands in for the S3 store the production
// provider writes the data bucket through.
type casBucketStore struct {
	b      *leasetest.VersionedBucket
	bucket string
}

func (s casBucketStore) Get(_ context.Context, key string) ([]byte, string, error) {
	body, ok := s.b.Current(s.bucket, key)
	if !ok {
		return nil, "", os.ErrNotExist
	}
	return body, s.b.LatestID(s.bucket, key), nil
}

func (s casBucketStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if ifMatch == "" {
		id, err := s.b.PutIfAbsent(ctx, s.bucket, key, body)
		if errors.Is(err, lease.ErrExists) {
			return "", release.ErrCASConflict
		}
		return id, err
	}
	if s.b.LatestID(s.bucket, key) != ifMatch {
		return "", release.ErrCASConflict
	}
	return s.b.Put(s.bucket, key, body), nil
}

func (s casBucketStore) List(ctx context.Context, prefix string) ([]string, error) {
	vs, err := s.b.ListVersions(ctx, s.bucket, prefix)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, v := range vs {
		if v.IsLatest && !v.DeleteMarker {
			out = append(out, v.Key)
		}
	}
	return out, nil
}

func (casBucketStore) ListDirs(context.Context, string) ([]string, error) {
	return nil, errors.New("test: not implemented")
}
func (casBucketStore) Delete(context.Context, string) error {
	return errors.New("test: not implemented")
}

// recordingReader records every exact-version read.
type recordingReader struct {
	lease.VersionedStore
	mu    sync.Mutex
	reads []string // key@version
}

func (r *recordingReader) GetVersion(ctx context.Context, bucket, key, version string) ([]byte, error) {
	r.mu.Lock()
	r.reads = append(r.reads, key+"@"+version)
	r.mu.Unlock()
	return r.VersionedStore.GetVersion(ctx, bucket, key, version)
}

type filesFixture struct {
	t      *testing.T
	claims release.Store
	bucket *leasetest.VersionedBucket
	data   *recordingReader
	files  casBucketStore
	dir    string
	f      *movesFiles
	// work is the start's captured work, modelled as the lock it holds
	// (the hostguard's): With takes mu, runs fn and releases it. workRan
	// counts the calls; workErr refuses them, every call when workErrAt is
	// 0, else the call with that 1-based index only (a crash at that
	// point).
	mu        sync.Mutex
	workRan   int
	workErr   error
	workErrAt int
}

const filesHost = "shop.example.com"

// filesCreated and filesInc are the fixture host's creation and incarnation
// (hostclaim.Incarnation), so its location is a valid origin tuple
// (instances/<host>/<incarnation>) for claim Plan 15's proof.
var (
	filesCreated = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
	filesInc     = func() string {
		inc, err := hostclaim.Incarnation("bab", "jam", filesHost, filesCreated)
		if err != nil {
			panic(err)
		}
		return inc
	}()
	filesLoc = "instances/" + filesHost + "/" + filesInc
)

func (x *filesFixture) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	x.mu.Lock()
	defer x.mu.Unlock()
	x.workRan++
	if x.workErr != nil && (x.workErrAt == 0 || x.workErrAt == x.workRan) {
		return x.workErr
	}
	return fn()
}

func newFilesFixture(t *testing.T) *filesFixture {
	t.Helper()
	// Claim Plan 15: the proof reads the claim's recorded admissions from
	// the claims store under both switches.
	t.Setenv("CARLOS_HOST_CLAIM", "1")
	t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
	b := leasetest.NewVersionedBucket()
	b.AddBucket("data")
	x := &filesFixture{t: t, claims: release.NewDirStore(t.TempDir()), bucket: b, data: &recordingReader{VersionedStore: b}, files: casBucketStore{b: b, bucket: "data"}, dir: t.TempDir()}
	x.f = &movesFiles{Files: x.files, Data: x.data, Bucket: "data", HomeRegion: "ie", Region: "uk", Box: "uk-box", DataDir: x.dir, Proof: ClaimOriginProof{Claims: x.claims, HomeRegion: "ie", Region: "uk"},
		EnvFile: flatClaimEnvFile(x.dir)}
	return x
}

func (x *filesFixture) envPath() string { return filepath.Join(x.dir, filesHost+".env") }

func (x *filesFixture) writeEnv(body string) {
	x.t.Helper()
	if err := os.WriteFile(x.envPath(), []byte(body), 0o600); err != nil {
		x.t.Fatal(err)
	}
}

// publish writes a manifest revision for generation gen through the
// ordinary publisher (a predecessor's session output) and returns it.
func (x *filesFixture) publish(gen string, stage func(p *hostfiles.Publisher)) hostfiles.Manifest {
	x.t.Helper()
	p, err := hostfiles.Open(context.Background(), x.files, movesFilesPrefix(filesLoc, gen))
	if err != nil {
		x.t.Fatal(err)
	}
	stage(p)
	if _, err := p.Commit(context.Background()); err != nil {
		x.t.Fatal(err)
	}
	m, found, err := hostfiles.Load(context.Background(), x.files, movesFilesPrefix(filesLoc, gen))
	if err != nil || !found {
		x.t.Fatal(err, found)
	}
	return m
}

func (x *filesFixture) manifest(gen string) (hostfiles.Manifest, bool) {
	x.t.Helper()
	m, found, err := hostfiles.Load(context.Background(), x.files, movesFilesPrefix(filesLoc, gen))
	if err != nil {
		x.t.Fatal(err)
	}
	return m, found
}

// servingObserved is a serving claim on uk-box whose holder is at gen with
// the cold admission chain given, newest last: each entry is
// "<generation>:<region>:<predecessor>:<predecessor region>". It is a
// hand-built claim: nothing in it is proven until observe stores it with
// recorded admissions (recordOrigins).
func servingObserved(gen string, chain ...string) hostclaim.Observed {
	h := &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + filesHost + ".json", Account: "bab", App: "jam", CreatedAt: filesCreated, Incarnation: filesInc, Location: filesLoc, Generation: gen, Session: 3, WriterBox: "uk-box", Stamp: 7}
	for i, c := range chain {
		p := strings.Split(c, ":")
		h.Admissions = append(h.Admissions, hostclaim.AdmissionRecord{Generation: p[0], Region: p[1], Predecessor: p[2], PredecessorRegion: p[3], Mode: "cold", Session: uint64(3 + i), Box: "uk-box"})
	}
	if n := len(h.Admissions); n > 0 {
		h.Session = h.Admissions[n-1].Session
	}
	return hostclaim.Observed{Host: filesHost, ETag: "e1", Claim: &hostclaim.Claim{V: 2, Host: filesHost, Gen: 9, Holder: h, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box"}}}
}

// recordOrigins records tuples as the claim's admitted origins, as the
// claim's CAS records an admission and the #250 files backfill records a
// live origin into an admitted prefix: one immutable node per tuple under
// control/host-origins/<host>/, chained from the claim's current head, and
// the head written on the stored claim. The claim must already be stored.
func recordOrigins(t *testing.T, claims release.Store, host string, tuples ...hostclaim.OriginTuple) {
	t.Helper()
	ctx := context.Background()
	o, err := hostclaim.Read(ctx, claims, host)
	if err != nil {
		t.Fatal(err)
	}
	if o.Absent {
		t.Fatalf("recordOrigins: %s has no stored claim", host)
	}
	c := o.Claim
	head := hostclaim.OriginHead{}
	if c.Origins != nil {
		head = *c.Origins
	}
	type node struct {
		V        int                   `json:"v"`
		Host     string                `json:"host"`
		Count    uint64                `json:"count"`
		Previous string                `json:"previous,omitempty"`
		Origin   hostclaim.OriginTuple `json:"origin"`
	}
	for _, tuple := range tuples {
		body, err := json.Marshal(node{V: 1, Host: host, Count: head.Count + 1, Previous: head.Digest, Origin: tuple})
		if err != nil {
			t.Fatal(err)
		}
		sum := sha256.New()
		sum.Write([]byte("carlos-admitted-origin-v1\x00"))
		sum.Write(body)
		digest := hex.EncodeToString(sum.Sum(nil))
		key := "control/host-origins/" + host + "/" + digest + ".json"
		if err := hostclaim.ValidateOriginObject(key, body); err != nil {
			t.Fatalf("recordOrigins: %v", err)
		}
		if _, err := claims.Put(ctx, key, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
			t.Fatal(err)
		}
		head = hostclaim.OriginHead{Digest: digest, Count: head.Count + 1}
	}
	c.Origins = &head
	raw, err := json.Marshal(c)
	if err != nil {
		t.Fatal(err)
	}
	key, err := hostclaim.Key(host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := claims.Put(ctx, key, raw, o.ETag); err != nil {
		t.Fatal(err)
	}
}

// observe stores the hand-built observation's claim in the fixture's claims
// store, records the named generations as admitted origins of the holder's
// incarnation and location in this region (the admitted generation among
// them, as its admission records it), and returns the stored observation,
// which is what Materialize is given and what the proof authenticates.
func (x *filesFixture) observe(o hostclaim.Observed, recorded ...string) hostclaim.Observed {
	x.t.Helper()
	ctx := context.Background()
	c := o.Claim
	if err := c.Validate(); err != nil {
		x.t.Fatalf("observe: %v", err)
	}
	key, err := hostclaim.Key(o.Host)
	if err != nil {
		x.t.Fatal(err)
	}
	etag := ""
	if stored, err := hostclaim.Read(ctx, x.claims, o.Host); err == nil && !stored.Absent {
		// A re-observation keeps the chain the store already records: the
		// head is the claim's, never a hand-built claim's.
		etag, c.Origins = stored.ETag, stored.Claim.Origins
	}
	raw, err := json.Marshal(c)
	if err != nil {
		x.t.Fatal(err)
	}
	if _, err := x.claims.Put(ctx, key, raw, etag); err != nil {
		x.t.Fatal(err)
	}
	tuples := make([]hostclaim.OriginTuple, 0, len(recorded))
	for _, gen := range recorded {
		tuples = append(tuples, hostclaim.OriginTuple{Incarnation: c.Holder.Incarnation, Location: c.Holder.Location, Region: "uk", Generation: gen})
	}
	if len(tuples) > 0 {
		recordOrigins(x.t, x.claims, o.Host, tuples...)
	}
	stored, err := hostclaim.Read(ctx, x.claims, o.Host)
	if err != nil {
		x.t.Fatal(err)
	}
	return stored
}

func launchInputs(newGen string) ClaimLaunchInputs {
	return ClaimLaunchInputs{Identity: hostwrite.Identity{Config: "sha256:c", Release: "sha256:r"}, Admission: hostclaim.Admission{Box: "uk-box", Mode: "cold", NewGeneration: newGen, Region: "uk"}}
}

func sha(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }

// The prepared manifest (spec §7.6, R19-1; claim rev 6.30 R18-1): one
// create-only manifest at sequence 0 under files/g-<P>/ in the location,
// every content object create-only at sequence 0 under the same prefix,
// the env file and the wake state as platform items, and the binding
// {location, P, op, stamp} returned for the op journal. A different build
// of the same generation is refused and leaves the manifest as it was.
func TestPreparedManifestIsCreateOnlySequenceZeroAndBound(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	for rel, body := range map[string]string{"a.txt": "A", "uploads/b.bin": "BB"} {
		if err := os.MkdirAll(filepath.Dir(filepath.Join(dir, rel)), 0o755); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(filepath.Join(dir, rel), []byte(body), 0o640); err != nil {
			t.Fatal(err)
		}
	}
	if err := os.Symlink("a.txt", filepath.Join(dir, "link")); err != nil {
		t.Fatal(err)
	}
	x.writeEnv("K=1\n")
	wake := hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{{Name: "tick", Kind: "cron", Spec: "* * * * *", NextAt: 5}}, NextDue: 5, Replication: "hot"}
	pm, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, wake)
	if err != nil {
		t.Fatal(err)
	}
	prefix := movesFilesPrefix(filesLoc, "mv1")
	if pm.Key != hostfiles.ManifestKey(prefix) || pm.Op != "mv1" || pm.Stamp != 7 || len(pm.SHA256) != 64 {
		t.Fatalf("binding: %+v", pm)
	}
	raw, ok := x.bucket.Current("data", pm.Key)
	if !ok || sha(raw) != pm.SHA256 {
		t.Fatalf("the manifest at %s is not the journalled bytes: %v", pm.Key, ok)
	}
	m, found := x.manifest("mv1")
	if !found || m.Seq != 0 {
		t.Fatalf("manifest found %v seq %d", found, m.Seq)
	}
	if err := hostfiles.ValidateManifest(m, []string{prefix}); err != nil {
		t.Fatal(err)
	}
	if len(m.Entries) != 2 || m.Entries["a.txt"].Mode != 0o640 || m.Entries["uploads/b.bin"].SHA256 != sha([]byte("BB")) {
		t.Fatalf("entries: %+v", m.Entries)
	}
	for _, e := range m.Entries {
		if e.Seq != 0 || !strings.HasPrefix(e.Object, prefix+"objects/0000000000000000-") {
			t.Fatalf("entry object: %+v", e)
		}
	}
	if e := m.Platform[hostfiles.EnvItem]; e.SHA256 != sha([]byte("K=1\n")) || e.Seq != 0 {
		t.Fatalf("env item: %+v", e)
	}
	ws, found, err := hostfiles.LoadWake(ctx, x.files, prefix, m, nil)
	if err != nil || !found || len(ws.Timers) != 1 || ws.NextDue != 5 || ws.Replication != "hot" {
		t.Fatalf("wake: %+v %v %v", ws, found, err)
	}
	// A different build of the same generation: refused, the manifest kept.
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("changed"), 0o640); err != nil {
		t.Fatal(err)
	}
	if _, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, wake); !errors.Is(err, hostfiles.ErrManifestExists) {
		t.Fatalf("second build: %v", err)
	}
	if again, _ := x.bucket.Current("data", pm.Key); sha(again) != pm.SHA256 {
		t.Fatal("the prepared manifest was replaced")
	}
}

// The build never turns missing inputs into an empty image (fix round 1,
// finding 4): the files directory the install placed must be there, and
// the env file must be there exactly when the install placed one (the
// archive's evidence, wantEnv). A tree that vanished, an expected env file
// that is gone, or an env file nobody placed, refuses before any write;
// an install whose archive carried no env file builds a manifest without
// the item.
func TestPreparedManifestRefusesMissingInputs(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	wake := hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}}
	build := func(gen string, wantEnv bool) (preparedManifest, error) {
		return buildPreparedManifest(ctx, x.bucket, "data", filesLoc, gen, gen, 7, dir, x.envPath(), wantEnv, wake)
	}
	written := func() int {
		vs, err := x.bucket.ListVersions(ctx, "data", strings.TrimSuffix(filesLoc, "/")+"/files/")
		if err != nil {
			t.Fatal(err)
		}
		return len(vs)
	}
	// No tree at all.
	x.writeEnv("K=1\n")
	if _, err := build("g1", true); err == nil {
		t.Fatal("built over a missing files directory")
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil {
		t.Fatal(err)
	}
	// The env file the install placed is gone.
	if err := os.Remove(x.envPath()); err != nil {
		t.Fatal(err)
	}
	if _, err := build("g1", true); err == nil {
		t.Fatal("built without the env file the install placed")
	}
	// An env file nobody placed.
	x.writeEnv("K=1\n")
	if _, err := build("g1", false); err == nil {
		t.Fatal("built over an env file the install did not place")
	}
	if n := written(); n != 0 {
		t.Fatalf("%d objects written by refused builds", n)
	}
	// An archive without an env file: a manifest without the item.
	if err := os.Remove(x.envPath()); err != nil {
		t.Fatal(err)
	}
	pm, err := build("g1", false)
	if err != nil {
		t.Fatal(err)
	}
	m, found := x.manifest("g1")
	if !found || len(m.Entries) != 1 {
		t.Fatalf("manifest: %+v %v", m, found)
	}
	if _, has := m.Platform[hostfiles.EnvItem]; has || pm.Key == "" {
		t.Fatalf("an env item for an archive without one: %+v", m.Platform)
	}
}

// The build never reads a file past the limit (fix round 1, finding 5):
// a file whose size is over hostfiles.MaxFileBytes is recorded as skipped
// from its stat alone, and every read is bounded at the limit plus one
// byte, so a file that grows while it is read is recorded as skipped
// rather than read whole. A sparse file stands in for the oversized one.
func TestPreparedBuildBoundsEveryRead(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil {
		t.Fatal(err)
	}
	big := filepath.Join(dir, "big.bin")
	if err := os.WriteFile(big, nil, 0o640); err != nil {
		t.Fatal(err)
	}
	if err := os.Truncate(big, hostfiles.MaxFileBytes+1); err != nil {
		t.Fatal(err)
	}
	x.writeEnv("K=1\n")
	reads := 0
	orig := readFileBounded
	readFileBounded = func(root *os.Root, rel string, limit int64) ([]byte, bool, error) {
		if path.Base(rel) == "big.bin" {
			reads++
		}
		return orig(root, rel, limit)
	}
	defer func() { readFileBounded = orig }()
	if _, err := x.f.BuildPrepared(ctx, filesLoc, "g1", "g1", 7, dir, x.envPath(), true, hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}}); err != nil {
		t.Fatal(err)
	}
	m, _ := x.manifest("g1")
	if e := m.Entries["big.bin"]; e.Skipped == "" || e.Object != "" {
		t.Fatalf("the oversized file was not recorded as skipped: %+v", e)
	}
	if reads != 0 {
		t.Fatalf("the oversized file was read %d times", reads)
	}
	if e := m.Entries["a.txt"]; e.SHA256 != sha([]byte("A")) {
		t.Fatalf("a.txt: %+v", e)
	}
	// The bounded read itself: a file larger than the limit is reported as
	// too large with at most limit+1 bytes read; one at the limit is read
	// whole.
	small := filepath.Join(x.dir, "small")
	if err := os.WriteFile(small, []byte("0123456789"), 0o600); err != nil {
		t.Fatal(err)
	}
	xroot, err := os.OpenRoot(x.dir)
	if err != nil {
		t.Fatal(err)
	}
	defer xroot.Close()
	body, tooLarge, err := orig(xroot, "small", 4)
	if err != nil || !tooLarge || len(body) > 5 {
		t.Fatalf("over the limit: %q %v %v", body, tooLarge, err)
	}
	body, tooLarge, err = orig(xroot, "small", 10)
	if err != nil || tooLarge || string(body) != "0123456789" {
		t.Fatalf("at the limit: %q %v %v", body, tooLarge, err)
	}
}

// A replay of the build (a crash after the objects, before or after the
// manifest) reuses every object it finds by digest and the manifest by
// key: the same binding, one version of each object, no new object.
func TestPreparedManifestReplayReusesObjects(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil {
		t.Fatal(err)
	}
	x.writeEnv("K=1\n")
	wake := hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}}
	first, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, wake)
	if err != nil {
		t.Fatal(err)
	}
	second, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, wake)
	if err != nil {
		t.Fatalf("replay: %v", err)
	}
	if first != second {
		t.Fatalf("replay binding differs: %+v %+v", first, second)
	}
	vs, err := x.bucket.ListVersions(ctx, "data", movesFilesPrefix(filesLoc, "mv1"))
	if err != nil {
		t.Fatal(err)
	}
	if len(vs) != 4 { // three objects (a.txt, env, wake) and the manifest, one version each
		t.Fatalf("%d versions under the prepared prefix after a replay: %+v", len(vs), vs)
	}
}

// A cold admission establishes its generation's files manifest from the
// newest valid manifest of its admitted predecessor chain, read by the
// exact version the listing names: here g2 (the predecessor) never
// published a manifest, so g1's current revision is the baseline. The copy
// keeps every object origin-qualified under g1, continues the sequence,
// and is create-only under g3.
func TestEstablishCopiesTheNewestValidManifestByExactVersion(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "old.txt", []byte("old"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	g1 := x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
		if err := p.StagePlatform(ctx, hostfiles.EnvItem, []byte("K=1\n")); err != nil {
			t.Fatal(err)
		}
	})
	if g1.Seq != 3 {
		t.Fatalf("g1 at seq %d", g1.Seq)
	}
	o := x.observe(servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk"), "g1", "g3")
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); err != nil {
		t.Fatal(err)
	}
	// One fence for the manifest's write (the unchanged env file takes no
	// new object) and the local effect.
	if x.workRan != 2 {
		t.Fatalf("work ran %d times", x.workRan)
	}
	m, found := x.manifest("g3")
	if !found {
		t.Fatal("no manifest under g3")
	}
	if m.Seq != 3 || len(m.Entries) != 2 || m.Entries["a.txt"].Object != g1.Entries["a.txt"].Object || m.Entries["old.txt"].Object != g1.Entries["old.txt"].Object {
		t.Fatalf("copied manifest: %+v", m)
	}
	if e := m.Platform[hostfiles.EnvItem]; e.Object != g1.Platform[hostfiles.EnvItem].Object {
		t.Fatalf("the unchanged env file took a new object: %+v", e)
	}
	want := hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")) + "@" + x.bucket.LatestID("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")))
	found = false
	for _, r := range x.data.reads {
		found = found || r == want
	}
	if !found {
		t.Fatalf("g1's manifest was not read at its exact latest version %s: %v", want, x.data.reads)
	}
	if _, has := x.manifest("g2"); has {
		t.Fatal("g2 gained a manifest")
	}
	// A repeat (a retry before the app executed) reads its own bytes back.
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); err != nil {
		t.Fatalf("repeat: %v", err)
	}
	if again, _ := x.manifest("g3"); again.Seq != 3 {
		t.Fatalf("the repeat replaced the manifest: seq %d", again.Seq)
	}
}

// Origins outlive admission retention (fix round 1, finding 6; platform-1e
// P2). AdmitWriter keeps eight admissions, while a copied manifest carries
// its oldest objects for as long as they are live: after g1..g10 the
// chain ends at g3 and the newest manifest still names g1's object. The
// rule a reference is held to is the claim's (P6-D3): within the
// incarnation's location in this region's bucket, under a generation's
// files prefix. The admitted chain selects the newest manifest; it does
// not bound what that manifest may name. An object outside the location,
// or under the location but not under a generation's files prefix, still
// refuses.
func TestEstablishKeepsOriginsPastAdmissionPruning(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	prev := x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
		if err := p.StagePlatform(ctx, hostfiles.EnvItem, []byte("K=1\n")); err != nil {
			t.Fatal(err)
		}
	})
	gens := []string{"g1"}
	for i := 2; i <= 10; i++ {
		gen := fmt.Sprintf("g%d", i)
		p, err := hostfiles.NewFromCopy(x.files, movesFilesPrefix(filesLoc, gen), prev, []string{movesFilesPrefix(filesLoc, "g1")})
		if err != nil {
			t.Fatal(err)
		}
		// Each copy carries its set as an establishment would: the previous
		// set, the chain then (g1 and the previous generation) and itself.
		p.SetOrigins(append(append([]string(nil), prev.Origins...), movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, gens[len(gens)-1]), movesFilesPrefix(filesLoc, gen)))
		if _, err := p.Commit(ctx); err != nil {
			t.Fatal(err)
		}
		prev, _ = x.manifest(gen)
		gens = append(gens, gen)
	}
	// The observation Materialize is given: g11 admitted, the holder at
	// g11, the eight newest admissions only: g4..g11.
	var chain []string
	for i := 4; i <= 11; i++ {
		chain = append(chain, fmt.Sprintf("g%d:uk:g%d:uk", i, i-1))
	}
	c := servingObserved("g11", chain...)
	if len(c.Claim.Holder.Admissions) != 8 {
		t.Fatalf("%d admissions", len(c.Claim.Holder.Admissions))
	}
	// Provenance comes from the claim (platform-1e's ruling, fix round 3;
	// claim Plan 15): the carried set is a hint, retained admissions are
	// never evidence, and every origin must be proven by the claim's
	// recorded admissions. g1 is retained-only (pruned from the chain and
	// never recorded), so the establishment refuses; once the claim records
	// g1 (its admission, or the #250 files backfill), it passes.
	o := x.observe(c, "g11")
	if err := x.f.Establish(ctx, o, launchInputs("g11"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("past pruning, g1 retained only: %v", err)
	}
	if _, found := x.manifest("g11"); found {
		t.Fatal("a manifest was written over an unproven ancestor")
	}
	o = x.observe(c, "g1")
	if err := x.f.Establish(ctx, o, launchInputs("g11"), x); err != nil {
		t.Fatalf("establish past pruning with the claim's record: %v", err)
	}
	m, found := x.manifest("g11")
	if !found || m.Entries["a.txt"].Object != movesFilesPrefix(filesLoc, "g1")+"objects/0000000000000001-"+sha([]byte("A")) {
		t.Fatalf("g11: %+v %v", m, found)
	}
	// The carried set (fix round 2, finding 3; Task 6b): only what the
	// entries reference, g1, plus the new generation. The hints for the
	// generations between, which nothing references, are dropped.
	want := []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g11")}
	if !reflect.DeepEqual(m.Origins, want) {
		t.Fatalf("g11 origins %v, want %v", m.Origins, want)
	}
	_ = gens
	// Under the location but not a generation's files prefix: refused.
	for name, object := range map[string]string{
		"another location":      "instances/other/files/g-g1/objects/0000000000000001-" + sha([]byte("A")),
		"not a files prefix":    strings.TrimSuffix(filesLoc, "/") + "/g-g1/objects/0000000000000001-" + sha([]byte("A")),
		"not a generation name": strings.TrimSuffix(filesLoc, "/") + "/files/g-/objects/0000000000000001-" + sha([]byte("A")),
	} {
		bad := prev
		bad.Entries = map[string]hostfiles.Entry{"a.txt": {Object: object, SHA256: sha([]byte("A")), Size: 1, Seq: 1}}
		x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g10")), mustJSON(t, bad))
		if err := x.f.Establish(ctx, o, launchInputs("g12"), x); !errors.Is(err, hostfiles.ErrOrigin) {
			t.Fatalf("%s: %v", name, err)
		}
	}
	if _, found := x.manifest("g12"); found {
		t.Fatal("a manifest was written over a foreign reference")
	}
}

// replacingStore replaces one key's object right after the first Get of
// it returns: another writer landing between a reader's check and its
// use.
type replacingStore struct {
	release.Store
	key     string
	replace func()
	once    sync.Once
}

func (s *replacingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	body, etag, err := s.Store.Get(ctx, key)
	if key == s.key && err == nil {
		s.once.Do(s.replace)
	}
	return body, etag, err
}

// preparedFixture builds a prepared manifest for mv1 with its binding
// journalled, the env file carrying the minted key, and the probe's
// observation and inputs.
func preparedFixture(t *testing.T) (*filesFixture, preparedManifest, hostclaim.Observed, ClaimLaunchInputs) {
	t.Helper()
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil {
		t.Fatal(err)
	}
	x.writeEnv("K=1\n")
	pm, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}})
	if err != nil {
		t.Fatal(err)
	}
	x.journal("mv1", pm)
	x.writeEnv("K=1\nCARLOS_INSTANCE_KEY=abc\n")
	o := servingObserved("mv1", "cur:ie:@root:ie", "mv1:uk:cur:ie")
	o.Claim.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "mv1", Phase: "probing", Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}
	o = x.observe(o, "mv1")
	in := launchInputs("mv1")
	in.Prepared = true
	return x, pm, o, in
}

// The probe's publisher is built from the bytes the establishment checked
// and the token captured with them, never from a second read (fix round
// 2, finding 2). Another writer replacing P's manifest between the check
// and the publication makes the compare-and-swap fail: nothing is
// published over bytes that were never compared with the journalled
// version, and the replacement stands as it was.
func TestEstablishPublishesOnlyOverTheBytesItChecked(t *testing.T) {
	x, pm, o, in := preparedFixture(t)
	ctx := context.Background()
	other := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 5, Entries: map[string]hostfiles.Entry{"b.txt": {Object: movesFilesPrefix(filesLoc, "mv1") + "objects/0000000000000005-" + sha([]byte("B")), SHA256: sha([]byte("B")), Size: 1, Seq: 5}}, Platform: map[string]hostfiles.Entry{}}
	replacement := mustJSON(t, other)
	x.f.Files = &replacingStore{Store: x.files, key: pm.Key, replace: func() { x.bucket.Put("data", pm.Key, replacement) }}
	if err := x.f.Establish(ctx, o, in, x); err == nil {
		t.Fatal("published over a manifest replaced after the check")
	}
	cur, _ := x.bucket.Current("data", pm.Key)
	if string(cur) != string(replacement) {
		t.Fatalf("the replacement was published over: %s", cur)
	}
	if x.workRan != 0 && x.workRan > 2 {
		t.Fatalf("the local effect ran after a refused publication (%d)", x.workRan)
	}
}

// A prepared replay is accepted only as exactly what this establishment
// would have written (fix round 2 finding 6, fix round 3): from the
// prepared baseline and the establishment's own operations, the env
// object at the sequence after the prepared one, the manifest at that
// sequence, the carried set the admitted one, the entries and other
// platform items unchanged, the adoption item gone; and the env object
// present at its key with the env's digest. Nothing is taken from the
// replay's own numbers. With correct Origins, an env entry under another
// location or generation, of the wrong size, torn, or at a fabricated
// sequence (99, or 2 where the establishment writes 1) whose object does
// not exist, is refused and left as it is. The genuine replay is accepted
// and writes nothing.
func TestEstablishRefusesAForeignEnvEntryOnPreparedReplay(t *testing.T) {
	x, pm, o, in := preparedFixture(t)
	ctx := context.Background()
	envBytes := []byte("K=1\nCARLOS_INSTANCE_KEY=abc\n")
	envSHA := sha(envBytes)
	raw, err := x.bucket.GetVersion(ctx, "data", pm.Key, pm.Version)
	if err != nil {
		t.Fatal(err)
	}
	var prepared hostfiles.Manifest
	if err := json.Unmarshal(raw, &prepared); err != nil {
		t.Fatal(err)
	}
	// The genuine establishment first: what this box writes, and the set it
	// carries, are the shape every replay is held to.
	if err := x.f.Establish(ctx, o, in, x); err != nil {
		t.Fatal(err)
	}
	genuine, _ := x.manifest("mv1")
	genuineRaw, _ := x.bucket.Current("data", pm.Key)
	if genuine.Seq != 1 || genuine.Platform[hostfiles.EnvItem].Seq != 1 || len(genuine.Origins) == 0 {
		t.Fatalf("the genuine establishment: %+v", genuine)
	}
	under := func(prefix string, seq int64) string { return fmt.Sprintf("%sobjects/%016d-%s", prefix, seq, envSHA) }
	for name, tc := range map[string]struct {
		seq   int64
		entry hostfiles.Entry
	}{
		"another location":         {1, hostfiles.Entry{Object: under("instances/other/files/g-mv1/", 1), SHA256: envSHA, Size: int64(len(envBytes)), Mode: 0o600, Seq: 1}},
		"another generation":       {1, hostfiles.Entry{Object: under(movesFilesPrefix(filesLoc, "cur"), 1), SHA256: envSHA, Size: int64(len(envBytes)), Mode: 0o600, Seq: 1}},
		"the wrong size":           {1, hostfiles.Entry{Object: under(movesFilesPrefix(filesLoc, "mv1"), 1), SHA256: envSHA, Size: 3, Mode: 0o600, Seq: 1}},
		"torn":                     {1, hostfiles.Entry{Object: under(movesFilesPrefix(filesLoc, "mv1"), 1), SHA256: envSHA, Size: int64(len(envBytes)), Mode: 0o600, Seq: 1, Torn: true}},
		"a fabricated sequence 99": {99, hostfiles.Entry{Object: under(movesFilesPrefix(filesLoc, "mv1"), 99), SHA256: envSHA, Size: int64(len(envBytes)), Mode: 0o600, Seq: 99}},
		"a fabricated sequence 2":  {2, hostfiles.Entry{Object: under(movesFilesPrefix(filesLoc, "mv1"), 2), SHA256: envSHA, Size: int64(len(envBytes)), Mode: 0o600, Seq: 2}},
	} {
		bumped := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: tc.seq, Entries: prepared.Entries, Platform: map[string]hostfiles.Entry{}, Origins: genuine.Origins}
		for k, v := range prepared.Platform {
			bumped.Platform[k] = v
		}
		bumped.Platform[hostfiles.EnvItem] = tc.entry
		body := mustJSON(t, bumped)
		x.bucket.Put("data", pm.Key, body)
		if err := x.f.Establish(ctx, o, in, x); err == nil {
			t.Fatalf("%s: accepted as this box's establishment", name)
		}
		if cur, _ := x.bucket.Current("data", pm.Key); string(cur) != string(body) {
			t.Fatalf("%s: the manifest was replaced: %s", name, cur)
		}
	}
	// The genuine bytes with their env object gone, or corrupt at its key
	// (other bytes, the entry's digest and size no longer its own): refused
	// too. The object is read and verified before a replay is accepted.
	envObject := genuine.Platform[hostfiles.EnvItem].Object
	x.bucket.Put("data", pm.Key, genuineRaw)
	x.bucket.BareDelete("data", envObject)
	if err := x.f.Establish(ctx, o, in, x); err == nil {
		t.Fatal("accepted a replay whose env object is gone")
	}
	x.bucket.Put("data", envObject, []byte("K=corrupt\nCARLOS_INSTANCE_KEY=abc"))
	if err := x.f.Establish(ctx, o, in, x); err == nil {
		t.Fatal("accepted a replay whose env object is corrupt")
	}
	x.bucket.Put("data", envObject, envBytes)
	// The genuine replay is accepted and writes nothing.
	x.bucket.Put("data", pm.Key, genuineRaw)
	seq1 := x.bucket.LatestID("data", pm.Key)
	if err := x.f.Establish(ctx, o, in, x); err != nil || x.bucket.LatestID("data", pm.Key) != seq1 {
		t.Fatalf("the genuine replay: %v, revision moved %v", err, x.bucket.LatestID("data", pm.Key) != seq1)
	}
}

// stallingStore holds one key's write until released, signalling that it
// is stalled: a slow S3 request.
type stallingStore struct {
	release.Store
	key     string
	stalled chan struct{}
	release chan struct{}
	once    sync.Once
}

func (s *stallingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key {
		s.once.Do(func() { close(s.stalled) })
		select {
		case <-s.release:
		case <-ctx.Done():
			return "", ctx.Err()
		}
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// Remote publication runs outside the start's work (fix round 1, finding
// 7; platform-1e P2): each bucket write is fenced by a short With (the
// selection still holds) and bounded by the request limit, and the lock
// is released before the request is made; With holds only the local
// effect. While the manifest's write stalls, another holder of the lock
// (BeginFence's stand-in) acquires it and proceeds; the establishment
// completes once the write returns.
func TestEstablishPublishesOutsideTheStartsWork(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	st := &stallingStore{Store: x.files, key: hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g2")), stalled: make(chan struct{}), release: make(chan struct{})}
	x.f.Files = st
	o := x.observe(servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk"), "g1", "g2")
	done := make(chan error, 1)
	go func() { done <- x.f.Establish(ctx, o, launchInputs("g2"), x) }()
	select {
	case <-st.stalled:
	case err := <-done:
		t.Fatalf("established before the manifest's write: %v", err)
	case <-time.After(10 * time.Second):
		t.Fatal("the manifest's write never began")
	}
	// The lock is free while the write stalls.
	acquired := make(chan error, 1)
	go func() {
		acquired <- x.With(ctx, hostwrite.Identity{Config: "c", Release: "r"}, func() error { return nil })
	}()
	select {
	case err := <-acquired:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the start's work is held across the stalled write: BeginFence would wait")
	}
	close(st.release)
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the establishment never completed")
	}
	if m, found := x.manifest("g2"); !found || m.Seq != 2 { // g1's one entry, then the env object
		t.Fatalf("g2: %+v %v", m, found)
	}
	// Two fences (the env object, the manifest), the probe above and the
	// local effect.
	if x.workRan != 4 {
		t.Fatalf("work ran %d times", x.workRan)
	}
}

// A never-admitted origin under the location is refused (fix round 2,
// finding 3; platform-1e's ruling): the location rule is syntax, not
// provenance. A manifest in an admitted generation naming an object
// under files/g-never/, which no admission and no carried set names, is
// refused with ErrOrigin, whether the manifest carries no set (the old
// format, judged by the chain alone) or a set that omits it; and a
// carried set is never widened by anything but the admitted chain.
func TestEstablishRefusesANeverAdmittedOrigin(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	g1 := x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	o := x.observe(servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk"), "g1", "g2")
	never := movesFilesPrefix(filesLoc, "never") + "objects/0000000000000001-" + sha([]byte("N"))
	x.bucket.Put("data", never, []byte("N"))
	for name, origins := range map[string][]string{"no carried set": nil, "a carried set without it": {movesFilesPrefix(filesLoc, "g1")}} {
		bad := g1
		bad.Entries = map[string]hostfiles.Entry{"a.txt": g1.Entries["a.txt"], "n.txt": {Object: never, SHA256: sha([]byte("N")), Size: 1, Seq: 1}}
		bad.Origins = origins
		x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")), mustJSON(t, bad))
		if err := x.f.Establish(ctx, o, launchInputs("g2"), x); !errors.Is(err, hostfiles.ErrOrigin) {
			t.Fatalf("%s: %v", name, err)
		}
		if _, found := x.manifest("g2"); found {
			t.Fatalf("%s: a manifest was written over a never-admitted origin", name)
		}
	}
	// Origins vouch for nothing by themselves (platform-1e's ruling): a
	// manifest that adds the never-granted origin to its own carried set
	// and names an object under it is refused. A carried hint that no entry
	// references is neither proven nor carried (Task 6b): the
	// establishment goes through and the hint does not survive it.
	bad := g1
	bad.Entries = map[string]hostfiles.Entry{"a.txt": g1.Entries["a.txt"], "n.txt": {Object: never, SHA256: sha([]byte("N")), Size: 1, Seq: 1}}
	bad.Origins = []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "never")}
	x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")), mustJSON(t, bad))
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("named and carried: %v", err)
	}
	if _, found := x.manifest("g2"); found {
		t.Fatal("a manifest was written over a self-vouched origin")
	}
	// The genuine manifest establishes, and the new set is exactly the
	// referenced origin and the new generation.
	x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")), mustJSON(t, g1))
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); err != nil {
		t.Fatal(err)
	}
	m, _ := x.manifest("g2")
	if want := []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g2")}; !reflect.DeepEqual(m.Origins, want) {
		t.Fatalf("g2 origins %v, want %v", m.Origins, want)
	}
	// A carried hint that no entry references (planted at g2, the newest
	// the chain from g3 reads): the establishment goes through and the
	// hint does not survive it.
	hint := m
	hint.Origins = []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g2"), movesFilesPrefix(filesLoc, "never")}
	x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g2")), mustJSON(t, hint))
	o3 := x.observe(servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk"), "g3")
	if err := x.f.Establish(ctx, o3, launchInputs("g3"), x); err != nil {
		t.Fatalf("carried only: %v", err)
	}
	m3, _ := x.manifest("g3")
	if want := []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g2"), movesFilesPrefix(filesLoc, "g3")}; !reflect.DeepEqual(m3.Origins, want) {
		t.Fatalf("g3 origins %v, want %v (the env object under g2 is referenced; the hint is not)", m3.Origins, want)
	}
}

// A legacy adopted location, with no incarnation component, gets the same
// treatment and is seeded at its first establishment: its old manifest
// carries no set and is judged by the admitted chain alone; the new
// manifest carries the chain and the new generation; a never-admitted
// origin under it is refused as anywhere else.
func TestEstablishSeedsALegacyLocationsOriginSet(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	legacy := "instances/" + filesHost
	loc := func(gen string) string { return movesFilesPrefix(legacy, gen) }
	p, err := hostfiles.Open(ctx, x.files, loc("g1"))
	if err != nil {
		t.Fatal(err)
	}
	if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
		t.Fatal(err)
	}
	if _, err := p.Commit(ctx); err != nil {
		t.Fatal(err)
	}
	old, _, _ := hostfiles.Load(ctx, x.files, loc("g1"))
	if old.Origins != nil {
		t.Fatalf("the legacy manifest carries a set: %v", old.Origins)
	}
	o := servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk")
	o.Claim.Holder.Location = legacy
	o = x.observe(o, "g1", "g2")
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); err != nil {
		t.Fatal(err)
	}
	m, found, err := hostfiles.Load(ctx, x.files, loc("g2"))
	if err != nil || !found {
		t.Fatal(err, found)
	}
	if want := []string{loc("g1"), loc("g2")}; !reflect.DeepEqual(m.Origins, want) || m.Entries["a.txt"].Object != old.Entries["a.txt"].Object {
		t.Fatalf("seeded: %+v", m)
	}
	bad := old
	bad.Entries = map[string]hostfiles.Entry{"n.txt": {Object: loc("never") + "objects/0000000000000001-" + sha([]byte("N")), SHA256: sha([]byte("N")), Size: 1, Seq: 1}}
	// Planted at g2, the newest manifest the chain from g3 reads.
	x.bucket.Put("data", hostfiles.ManifestKey(loc("g2")), mustJSON(t, bad))
	o3 := servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk")
	o3.Claim.Holder.Location = legacy
	o3 = x.observe(o3, "g3")
	if err := x.f.Establish(ctx, o3, launchInputs("g3"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("never admitted under a legacy location: %v", err)
	}
}

// answeringProof is a claim-backed origin proof with an answer per
// generation: proven, not proven, or an error; a generation it does not
// name is not proven.
type answeringProof struct {
	proven map[string]bool
	errs   map[string]error
}

func (a answeringProof) Proven(_ context.Context, _ hostclaim.Observed, _ hostclaim.Admission, generations []string) (map[string]bool, error) {
	out := map[string]bool{}
	for _, generation := range generations {
		if err := a.errs[generation]; err != nil {
			return nil, err
		}
		out[generation] = a.proven[generation]
	}
	return out, nil
}

// movesOriginProof is the only route to origin proof (fix round 4): the
// chain's prefixes and the prepared generation go through Proven like
// every other origin, so a different proof (claim Plan 15's) has the final
// say. A proof that answers false, or an error, for a chain-prefix origin
// or for the prepared generation's origin refuses the establishment and
// nothing is written.
func TestEstablishProvesEveryOriginThroughTheProof(t *testing.T) {
	ctx := context.Background()
	for name, mk := range map[string]func(gen string) answeringProof{
		"not proven": func(gen string) answeringProof { return answeringProof{proven: map[string]bool{}} },
		"an error": func(gen string) answeringProof {
			return answeringProof{errs: map[string]error{gen: errors.New("the record is unreadable")}}
		},
	} {
		t.Run("a chain prefix, "+name, func(t *testing.T) {
			x := newFilesFixture(t)
			x.writeEnv("K=1\n")
			x.publish("g1", func(p *hostfiles.Publisher) {
				if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
					t.Fatal(err)
				}
			})
			o := servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk")
			proof := mk("g1")
			proof.proven = map[string]bool{"g2": true}
			x.f.Proof = proof
			err := x.f.Establish(ctx, o, launchInputs("g2"), x)
			if err == nil || (name == "not proven" && !errors.Is(err, hostfiles.ErrOrigin)) {
				t.Fatalf("a chain prefix the proof does not prove: %v", err)
			}
			if _, found := x.manifest("g2"); found {
				t.Fatal("a manifest was written over an unproven chain prefix")
			}
			// The new generation itself is proven through the proof too.
			x.f.Proof = answeringProof{proven: map[string]bool{"g1": true}}
			if err := x.f.Establish(ctx, o, launchInputs("g2"), x); !errors.Is(err, hostfiles.ErrOrigin) {
				t.Fatalf("the new generation the proof does not prove: %v", err)
			}
			x.f.Proof = answeringProof{proven: map[string]bool{"g1": true, "g2": true}}
			if err := x.f.Establish(ctx, o, launchInputs("g2"), x); err != nil {
				t.Fatalf("everything proven: %v", err)
			}
		})
		t.Run("the prepared generation, "+name, func(t *testing.T) {
			x, pm, o, in := preparedFixture(t)
			proof := mk("mv1")
			proof.proven = map[string]bool{"cur": true}
			x.f.Proof = proof
			err := x.f.Establish(ctx, o, in, x)
			if err == nil || (name == "not proven" && !errors.Is(err, hostfiles.ErrOrigin)) {
				t.Fatalf("the prepared generation the proof does not prove: %v", err)
			}
			if x.bucket.LatestID("data", pm.Key) != pm.Version {
				t.Fatal("the prepared manifest was published over")
			}
			x.f.Proof = answeringProof{proven: map[string]bool{"cur": true, "mv1": true}}
			if err := x.f.Establish(ctx, o, in, x); err != nil {
				t.Fatalf("everything proven: %v", err)
			}
		})
	}
}

// The proof never accepts a bare Admission.NewGeneration supplied on its
// own (platform-1e's condition, fix round 4; claim Plan 15): a generation
// is proven only when the bound observation's claim records its
// admission, as the post-admission observation Materialize is given does.
// A free-standing generation, named by the admission but recorded by no
// observation, is refused at the proof and by Establish, and nothing is
// written; an observation of no claim is an error.
func TestClaimProofRefusesAFreeStandingGeneration(t *testing.T) {
	ctx := context.Background()
	x := newFilesFixture(t)
	x.writeEnv("K=1\n")
	x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	proof := ClaimOriginProof{Claims: x.claims, HomeRegion: "ie", Region: "uk"}
	before := x.observe(servingObserved("g1", "g1:uk:@root:uk"), "g1") // the observation before g2's admission
	in := hostclaim.Admission{Box: "uk-box", Mode: "cold", NewGeneration: "g2", Region: "uk"}
	if got, err := proof.Proven(ctx, before, in, []string{"g1", "g2"}); err != nil || !got["g1"] || got["g2"] {
		t.Fatalf("a free-standing generation: %v %v", got, err)
	}
	if _, err := proof.Proven(ctx, hostclaim.Observed{Host: filesHost, Absent: true}, in, []string{"g2"}); err == nil {
		t.Fatal("proven with no claim")
	}
	if err := x.f.Establish(ctx, before, launchInputs("g2"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("establish with an observation that does not record the admission: %v", err)
	}
	if _, found := x.manifest("g2"); found {
		t.Fatal("a manifest was written for a free-standing generation")
	}
	after := x.observe(servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk"), "g2") // the CAS result the writer's admission returns
	if err := x.f.Establish(ctx, after, launchInputs("g2"), x); err != nil {
		t.Fatalf("establish with the bound observation: %v", err)
	}
}

// countingProof records every Proven call and the set it was asked about,
// answering from proven.
type countingProof struct {
	proven map[string]bool
	calls  [][]string
}

func (c *countingProof) Proven(_ context.Context, _ hostclaim.Observed, _ hostclaim.Admission, generations []string) (map[string]bool, error) {
	c.calls = append(c.calls, append([]string(nil), generations...))
	out := map[string]bool{}
	for _, generation := range generations {
		out[generation] = c.proven[generation]
	}
	return out, nil
}

// Only referenced origins are proven and carried (Task 6b, the moves side
// of claim Plan 15's review): the origins the live manifest's entries
// name, plus the one generation the admission opens (round 4's binding,
// one call). A carried hint nothing references is neither proven nor
// carried forward, so old hints do not accumulate and the proof's cost
// does not grow with the history; a referenced origin is still proven and
// refused when the proof says no; and the count of Proven calls is the
// distinct referenced origins plus the admitted generation, each asked
// once.
func TestEstablishProvesAndCarriesOnlyReferencedOrigins(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	// g5's object, referenced by the live manifest at g1 through a carried
	// entry; g7 is a hint nothing references and nothing can prove.
	g5obj := movesFilesPrefix(filesLoc, "g5") + "objects/0000000000000001-" + sha([]byte("F"))
	x.bucket.Put("data", g5obj, []byte("F"))
	live := x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	live.Entries["f.txt"] = hostfiles.Entry{Object: g5obj, SHA256: sha([]byte("F")), Size: 1, Seq: 1}
	live.Origins = []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g5"), movesFilesPrefix(filesLoc, "g7")}
	x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g1")), mustJSON(t, live))
	o := servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk")
	// The hint g7 is not proven: the establishment still goes through, and
	// g7 is not carried; g1, g5 and the admitted g2 are, each asked once.
	proof := &countingProof{proven: map[string]bool{"g1": true, "g2": true, "g5": true}}
	x.f.Proof = proof
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); err != nil {
		t.Fatalf("an unreferenced hint blocked the establishment: %v", err)
	}
	m, found := x.manifest("g2")
	want := []string{movesFilesPrefix(filesLoc, "g1"), movesFilesPrefix(filesLoc, "g2"), movesFilesPrefix(filesLoc, "g5")}
	if !found || !reflect.DeepEqual(m.Origins, want) {
		t.Fatalf("g2 origins %v, want %v", m.Origins, want)
	}
	// One Proven(set) call per Establish (Plan 15): the deduplicated
	// referenced origins plus the admitted generation, in one batch (the
	// set's order is the caller's; the prefixes are what get sorted).
	if len(proof.calls) != 1 {
		t.Fatalf("Proven calls %v, want one call over the referenced origins and the admitted generation", proof.calls)
	}
	asked := append([]string(nil), proof.calls[0]...)
	sort.Strings(asked)
	if !reflect.DeepEqual(asked, []string{"g1", "g2", "g5"}) {
		t.Fatalf("Proven asked %v, want the referenced origins and the admitted generation", proof.calls[0])
	}
	// A referenced origin the proof refuses still refuses the whole.
	x.f.Proof = &countingProof{proven: map[string]bool{"g1": true, "g2": true, "g3": true}}
	if err := x.f.Establish(ctx, servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk"), launchInputs("g3"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("a referenced origin the proof refuses: %v", err)
	}
	if _, found := x.manifest("g3"); found {
		t.Fatal("a manifest was written over a refused referenced origin")
	}
}

// recoveryObserved is the source box's observation after a recovery
// admission: the holder at the fresh recovery generation R, admitted after
// the target's admission T in the other region, under a restoring op (or
// a reverting one) whose journal names the source's generation src_gen,
// as the flip wrote it.
func recoveryObserved(kind, srcGen, target, recovered string) hostclaim.Observed {
	o := servingObserved(recovered, "src:uk:@root:uk", target+":ie:src:uk", recovered+":uk:"+target+":ie")
	if srcGen == "" {
		o.Claim.Holder.Admissions = o.Claim.Holder.Admissions[1:] // a legacy source: no admission of its own
	}
	o.Claim.Op = hostclaim.Op{Kind: kind, ID: target, Phase: hostclaim.PhaseProbing, Exec: "uk-box", Participants: []string{"uk-box", "ie-box"},
		Intent:  map[string]string{"to": "ie", "source": "uk-box", "target": "ie-box", "rollback": "rb1"},
		Journal: map[string]string{hostclaim.JournalSrcGen: srcGen, hostclaim.JournalSourceStamp: "15", hostclaim.JournalCutSession: "3"}}
	return o
}

// A recovery admission's files baseline is the recorded source restore
// (whole-branch fixes, item 3, as re-ruled with platform-1e): for a
// restoring op whose own lineage holds no manifest, the chain walk starts
// at the journalled src_gen, pinned as the database's source restore is,
// because the admission's predecessor is the target's generation in the
// other region and ends the ordinary walk at once. The origin proof still
// goes only through movesOriginProof with the bound observation: src_gen
// is proven by the claim's journal, and a proof that refuses it refuses
// the whole. Clean, declared-loss and fenced returns establish from
// src_gen's manifest; a revert, whose holder never left the source,
// reaches it through its own lineage; a legacy source (src_gen "") has no
// files generation and refuses.
func TestRecoveryEstablishesFromTheRecordedSourceGeneration(t *testing.T) {
	ctx := context.Background()
	for name, shape := range map[string]func(o *hostclaim.Observed){
		"a clean return": func(*hostclaim.Observed) {},
		"a declared-loss return": func(o *hostclaim.Observed) {
			o.Claim.Op.Journal["loss.l1"] = `{"kind":"move-target-lost","target":"ie-box"}`
			o.Claim.LossRefs, o.Claim.Lost = []string{"l1"}, []string{"ie-box"}
		},
		"a fenced return": func(o *hostclaim.Observed) {
			o.Claim.Op.Journal["fence.ie-box"] = `{"enrollment":"enr","key":"control/box-fences/enr.json","sha256":"ab"}`
		},
		"a revert": func(o *hostclaim.Observed) {
			// A revert's holder never left the source: its admission's
			// predecessor is src in this region, so it follows its own
			// lineage with no bridge (platform-1e's framing).
			o.Claim.Op.Kind = hostclaim.KindReverting
			delete(o.Claim.Op.Intent, "rollback")
			o.Claim.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "src", Region: "uk", Predecessor: "@root", PredecessorRegion: "uk", Mode: "cold", Session: 3, Box: "uk-box"},
				{Generation: "r1", Region: "uk", Predecessor: "src", PredecessorRegion: "uk", Mode: "cold", Session: 3, Box: "uk-box"}}
		},
	} {
		t.Run(name, func(t *testing.T) {
			x := newFilesFixture(t)
			x.writeEnv("K=1\n")
			src := x.publish("src", func(p *hostfiles.Publisher) {
				if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
					t.Fatal(err)
				}
			})
			o := recoveryObserved(hostclaim.KindRestoring, "src", "mv1", "r1")
			shape(&o)
			o = x.observe(o, "src", "r1")
			if err := x.f.Establish(ctx, o, launchInputs("r1"), x); err != nil {
				t.Fatalf("establish from src_gen: %v", err)
			}
			m, found := x.manifest("r1")
			if !found || m.Entries["a.txt"].Object != src.Entries["a.txt"].Object || !reflect.DeepEqual(m.Origins, []string{movesFilesPrefix(filesLoc, "r1"), movesFilesPrefix(filesLoc, "src")}) {
				t.Fatalf("r1: %+v %v", m, found)
			}
			// The proof has the final say over src_gen.
			y := newFilesFixture(t)
			y.writeEnv("K=1\n")
			y.publish("src", func(p *hostfiles.Publisher) {
				if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
					t.Fatal(err)
				}
			})
			y.f.Proof = answeringProof{proven: map[string]bool{"r1": true}}
			if err := y.f.Establish(ctx, o, launchInputs("r1"), y); !errors.Is(err, hostfiles.ErrOrigin) {
				t.Fatalf("src_gen the proof refuses: %v", err)
			}
		})
	}
	// A legacy source: src_gen "" names no files generation; nothing to
	// establish from, and never a bootstrap.
	x := newFilesFixture(t)
	x.writeEnv("K=1\n")
	if err := x.f.Establish(ctx, recoveryObserved(hostclaim.KindRestoring, "", "mv1", "r1"), launchInputs("r1"), x); !errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("a legacy source: %v", err)
	}
}

// recoveryRestartObserved is the source box's observation after a later
// recovery admission: the holder at the fresh generation `next`, admitted
// after the recovery generation `prev` in this region (the prepared
// rollback generation, or an earlier recovery admission), itself admitted
// after the target's T in the other region.
func recoveryRestartObserved(kind, prev, next string) hostclaim.Observed {
	o := servingObserved(next, "src:uk:@root:uk", "mv1:ie:src:uk", prev+":uk:mv1:ie", next+":uk:"+prev+":uk")
	o.Claim.Op = hostclaim.Op{Kind: kind, ID: "mv1", Phase: hostclaim.PhaseProbing, Exec: "uk-box", Participants: []string{"uk-box", "ie-box"},
		Intent:  map[string]string{"to": "ie", "source": "uk-box", "target": "ie-box", "rollback": "rb1"},
		Journal: map[string]string{hostclaim.JournalSrcGen: "src", hostclaim.JournalSourceStamp: "15", hostclaim.JournalCutSession: "3"}}
	if kind == hostclaim.KindReverting {
		delete(o.Claim.Op.Intent, "rollback")
	}
	return o
}

// src_gen only bridges the missing source baseline (the re-review's
// ruling on item 3, mirroring the database selector): once a recovery
// manifest exists, the prepared rollback generation's or a later recovery
// publication's, the newest authorized recovery manifest wins through the
// normal chain and origin proof. A restart after the dirty return's
// prepared rollback manifest publishes the rollback's contents (B), not
// the source's (A), even with the source manifest absent; a clean return
// and a revert after a newer recovery publication take the newer one; a
// first admission with no recovery manifest still bridges from src_gen.
func TestRecoveryRestartTakesTheNewestRecoveryManifest(t *testing.T) {
	ctx := context.Background()
	publish := func(x *filesFixture, gen, body string) hostfiles.Manifest {
		return x.publish(gen, func(p *hostfiles.Publisher) {
			if err := p.Stage(ctx, "a.txt", []byte(body), 0o600, false); err != nil {
				t.Fatal(err)
			}
		})
	}
	t.Run("a restart after the prepared rollback manifest publishes B", func(t *testing.T) {
		for name, withSource := range map[string]bool{"source present": true, "source absent": false} {
			t.Run(name, func(t *testing.T) {
				x := newFilesFixture(t)
				x.writeEnv("K=1\n")
				if withSource {
					publish(x, "src", "A")
				}
				rb := publish(x, "rb1", "B")
				o := x.observe(recoveryRestartObserved(hostclaim.KindRestoring, "rb1", "r2"), "rb1", "r2")
				if err := x.f.Establish(ctx, o, launchInputs("r2"), x); err != nil {
					t.Fatalf("the restart: %v", err)
				}
				m, found := x.manifest("r2")
				if !found || m.Entries["a.txt"].Object != rb.Entries["a.txt"].Object {
					t.Fatalf("r2 took %+v, want the rollback's B (%s)", m.Entries["a.txt"], rb.Entries["a.txt"].Object)
				}
			})
		}
	})
	for name, kind := range map[string]string{"a clean return": hostclaim.KindRestoring, "a revert": hostclaim.KindReverting} {
		t.Run(name+" after a newer recovery publication", func(t *testing.T) {
			x := newFilesFixture(t)
			x.writeEnv("K=1\n")
			publish(x, "src", "A")
			newer := publish(x, "r1", "C") // the earlier recovery admission's publication
			o := x.observe(recoveryRestartObserved(kind, "r1", "r2"), "r1", "r2")
			if err := x.f.Establish(ctx, o, launchInputs("r2"), x); err != nil {
				t.Fatalf("the restart: %v", err)
			}
			m, found := x.manifest("r2")
			if !found || m.Entries["a.txt"].Object != newer.Entries["a.txt"].Object {
				t.Fatalf("r2 took %+v, want the newer recovery's C", m.Entries["a.txt"])
			}
		})
	}
	t.Run("a dirty return's later admissions follow the rollback lineage", func(t *testing.T) {
		// The second cold admission restores from the rollback generation,
		// the third from its certified descendant (the second's publication).
		x := newFilesFixture(t)
		x.writeEnv("K=1\n")
		publish(x, "src", "A")
		rb := publish(x, "rb1", "B")
		o := recoveryRestartObserved(hostclaim.KindRestoring, "rb1", "r2")
		o.Claim.Op.Journal[moves.JournalReturnSeal] = "sealed"
		o = x.observe(o, "rb1", "r2")
		if err := x.f.Establish(ctx, o, launchInputs("r2"), x); err != nil {
			t.Fatalf("the second admission: %v", err)
		}
		if m, _ := x.manifest("r2"); m.Entries["a.txt"].Object != rb.Entries["a.txt"].Object {
			t.Fatalf("r2 took %+v, want the rollback's B", m.Entries["a.txt"])
		}
		r2 := publish(x, "r2", "D") // the second admission's own publication
		o3 := servingObserved("r3", "src:uk:@root:uk", "mv1:ie:src:uk", "rb1:uk:mv1:ie", "r2:uk:rb1:uk", "r3:uk:r2:uk")
		o3.Claim.Op = o.Claim.Op
		o3 = x.observe(o3, "r3")
		if err := x.f.Establish(ctx, o3, launchInputs("r3"), x); err != nil {
			t.Fatalf("the third admission: %v", err)
		}
		if m, _ := x.manifest("r3"); m.Entries["a.txt"].Object != r2.Entries["a.txt"].Object {
			t.Fatalf("r3 took %+v, want the descendant's D", m.Entries["a.txt"])
		}
	})
	t.Run("a dirty return never bridges to the source", func(t *testing.T) {
		// A dirty return restores the rollback generation, never the stopped
		// source: with no manifest in its lineage it refuses rather than
		// publishing the source's older files.
		x := newFilesFixture(t)
		x.writeEnv("K=1\n")
		publish(x, "src", "A")
		o := recoveryRestartObserved(hostclaim.KindRestoring, "rb1", "r2")
		o.Claim.Op.Journal[moves.JournalReturnSeal] = "sealed"
		o = x.observe(o, "src", "r2")
		if err := x.f.Establish(ctx, o, launchInputs("r2"), x); !errors.Is(err, hostfiles.ErrNoEvidence) {
			t.Fatalf("a dirty return without its rollback manifest: %v", err)
		}
		if _, found := x.manifest("r2"); found {
			t.Fatal("a dirty return published the source's files")
		}
	})
	t.Run("a first admission with no recovery manifest bridges from src_gen", func(t *testing.T) {
		x := newFilesFixture(t)
		x.writeEnv("K=1\n")
		src := publish(x, "src", "A")
		o := x.observe(recoveryObserved(hostclaim.KindRestoring, "src", "mv1", "r1"), "src", "r1")
		if err := x.f.Establish(ctx, o, launchInputs("r1"), x); err != nil {
			t.Fatalf("the first admission: %v", err)
		}
		if m, found := x.manifest("r1"); !found || m.Entries["a.txt"].Object != src.Entries["a.txt"].Object {
			t.Fatalf("r1 took %+v, want the source's A", m.Entries["a.txt"])
		}
	})
}

// The proven set and the carried set order the same way (Astra's finding
// on the Plan 15 switch): generation names and their files prefixes do
// not sort alike ("mv1" before "mv1-old", but "g-mv1-old/" before
// "g-mv1/"), so the prefixes, not the names, are sorted everywhere the
// sets are compared. A prepared manifest for mv1 that also references an
// object under mv1-old (a same-region retained reference) is established
// once; a crash before the local effect leaves the key at sequence 1
// with the carried set; the genuine replay is accepted and writes nothing.
func TestPreparedReplayOrdersOriginsLikeTheCarriedSet(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	oldObj := movesFilesPrefix(filesLoc, "mv1-old") + "objects/0000000000000001-" + sha([]byte("O"))
	x.bucket.Put("data", oldObj, []byte("O"))
	newObj := movesFilesPrefix(filesLoc, "mv1") + "objects/0000000000000000-" + sha([]byte("A"))
	x.bucket.Put("data", newObj, []byte("A"))
	envObj := movesFilesPrefix(filesLoc, "mv1") + "objects/0000000000000000-" + sha([]byte("K=1\n"))
	x.bucket.Put("data", envObj, []byte("K=1\n"))
	prepared := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 0,
		Entries:  map[string]hostfiles.Entry{"a.txt": {Object: newObj, SHA256: sha([]byte("A")), Size: 1, Mode: 0o600, Seq: 0}, "old.txt": {Object: oldObj, SHA256: sha([]byte("O")), Size: 1, Mode: 0o600, Seq: 1}},
		Platform: map[string]hostfiles.Entry{hostfiles.EnvItem: {Object: envObj, SHA256: sha([]byte("K=1\n")), Size: 4, Mode: 0o600, Seq: 0}},
		Origins:  []string{movesFilesPrefix(filesLoc, "mv1-old"), movesFilesPrefix(filesLoc, "mv1")}}
	raw := mustJSON(t, prepared)
	key := hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "mv1"))
	version := x.bucket.Put("data", key, raw)
	x.journal("mv1", preparedManifest{Key: key, SHA256: sha(raw), Version: version, Op: "mv1", Stamp: 7})
	x.writeEnv("K=1\nCARLOS_INSTANCE_KEY=abc\n")
	o := servingObserved("mv1", "cur:ie:@root:ie", "mv1:uk:cur:ie")
	o.Claim.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "mv1", Phase: "probing", Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}
	o = x.observe(o, "mv1", "mv1-old")
	in := launchInputs("mv1")
	in.Prepared = true
	x.workErr, x.workErrAt = errors.New("crash before the local effect"), 3
	if err := x.f.Establish(ctx, o, in, x); err == nil {
		t.Fatal("the crash was not reported")
	}
	x.workErr, x.workErrAt = nil, 0
	m, found := x.manifest("mv1")
	if !found || m.Seq != 1 {
		t.Fatalf("after the crashed attempt: %+v %v", m, found)
	}
	seq1 := x.bucket.LatestID("data", key)
	if err := x.f.Establish(ctx, o, in, x); err != nil {
		t.Fatalf("the genuine replay: %v", err)
	}
	if x.bucket.LatestID("data", key) != seq1 {
		t.Fatal("the replay wrote a new revision")
	}
}

// The instance key rule (O1'): the manifest Establish leaves carries the
// env file with exactly the bytes on disk when it runs, staged and
// committed before it returns, and no adoption item; the local adoption
// record is removed under the start's work. Establish never writes the
// env file. A work that refuses leaves everything as it was.
func TestEstablishCarriesTheEnvBytesAndDropsTheAdoptionItem(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.StagePlatform(ctx, hostfiles.EnvItem, []byte("K=old\n")); err != nil {
			t.Fatal(err)
		}
		if err := p.StagePlatform(ctx, instancekey.AdoptionItem, []byte(`{"v":2}`)); err != nil {
			t.Fatal(err)
		}
	})
	x.writeEnv("K=new\nCARLOS_INSTANCE_KEY=abc\n")
	adoption := filepath.Join(x.dir, filesHost+instancekey.AdoptionSuffix)
	if err := os.WriteFile(adoption, []byte(`{"v":2}`), 0o600); err != nil {
		t.Fatal(err)
	}
	o := x.observe(servingObserved("g2", "g1:uk:@root:uk", "g2:uk:g1:uk"), "g1", "g2")
	x.workErr = hostbox.ErrChanged
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("refused work: %v", err)
	}
	if _, found := x.manifest("g2"); found {
		t.Fatal("a refused work wrote the manifest")
	}
	if _, err := os.Stat(adoption); err != nil {
		t.Fatal("a refused work removed the adoption record")
	}
	x.workErr = nil
	if err := x.f.Establish(ctx, o, launchInputs("g2"), x); err != nil {
		t.Fatal(err)
	}
	m, found := x.manifest("g2")
	if !found {
		t.Fatal("no manifest under g2")
	}
	e := m.Platform[hostfiles.EnvItem]
	if e.SHA256 != sha([]byte("K=new\nCARLOS_INSTANCE_KEY=abc\n")) || !strings.HasPrefix(e.Object, movesFilesPrefix(filesLoc, "g2")+"objects/") {
		t.Fatalf("env item: %+v", e)
	}
	got, err := hostfiles.ReadPlatform(ctx, x.files, movesFilesPrefix(filesLoc, "g2"), m, hostfiles.EnvItem, nil)
	if err != nil || string(got) != "K=new\nCARLOS_INSTANCE_KEY=abc\n" {
		t.Fatalf("env bytes: %q %v", got, err)
	}
	if _, has := m.Platform[instancekey.AdoptionItem]; has {
		t.Fatal("the adoption item was carried into the new generation")
	}
	if _, err := os.Stat(adoption); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the local adoption record remains: %v", err)
	}
	if disk, _ := os.ReadFile(x.envPath()); string(disk) != "K=new\nCARLOS_INSTANCE_KEY=abc\n" {
		t.Fatalf("Establish wrote the env file: %q", disk)
	}
	// Without an env file on disk there is nothing to carry: an error, and
	// no manifest.
	if err := os.Remove(x.envPath()); err != nil {
		t.Fatal(err)
	}
	if err := x.f.Establish(ctx, servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk"), launchInputs("g3"), x); err == nil {
		t.Fatal("established without an env file")
	}
	if _, found := x.manifest("g3"); found {
		t.Fatal("a manifest was written without the env file")
	}
}

// Missing or malformed data is an error, never an empty-files bootstrap
// (claim §6.4a): a chain with no manifest anywhere, with no pin evidence
// or with evidence of published files, refuses; a malformed newest
// manifest refuses rather than falling back to an older one; a chain that
// leaves this region is not followed into another region's bucket. Nothing
// is written in any of these.
func TestEstablishNeverBootstrapsEmpty(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	o := servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk")
	// No manifest anywhere, no evidence.
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); !errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("no manifest, no pin: %v", err)
	}
	// A pin that evidences published files is not evidence of none.
	x.f.Evidence = func(context.Context, hostclaim.Observed) (hostfiles.PinnedFiles, bool, error) {
		return hostfiles.PinnedFiles{SHA256: "ab", Retained: "retained/ab"}, true, nil
	}
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); !errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("no manifest, published pin: %v", err)
	}
	// A pin reader that fails is an error, not absence.
	x.f.Evidence = func(context.Context, hostclaim.Observed) (hostfiles.PinnedFiles, bool, error) {
		return hostfiles.PinnedFiles{}, false, errors.New("pin unreadable")
	}
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); err == nil || errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("unreadable pin: %v", err)
	}
	x.f.Evidence = nil
	// The newest manifest is malformed: refused, not skipped.
	x.publish("g1", func(p *hostfiles.Publisher) {
		if err := p.Stage(ctx, "a.txt", []byte("A"), 0o600, false); err != nil {
			t.Fatal(err)
		}
	})
	x.bucket.Put("data", hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "g2")), []byte(`{"v":2,"seq":1,"entries":{"x":{"object":"elsewhere/objects/0000000000000001-`+sha(nil)+`","sha256":"`+sha(nil)+`","seq":1}},"platform":{}}`))
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("malformed newest manifest: %v", err)
	}
	// The predecessor lives in another region: its manifest is not in this
	// bucket and the chain is not followed there.
	far := servingObserved("g6", "g5:uk:g1:ie", "g6:uk:g5:uk")
	if err := x.f.Establish(ctx, far, launchInputs("g6"), x); !errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("predecessor in another region: %v", err)
	}
	for _, gen := range []string{"g3", "g6"} {
		if _, found := x.manifest(gen); found {
			t.Fatalf("a manifest was written under %s", gen)
		}
	}
	if x.workRan != 0 {
		t.Fatalf("work ran %d times with nothing to establish", x.workRan)
	}
}

// The one empty start besides the claim's own bootstrap (in.Empty after
// EmptyAuthority): the pin evidences a host that never published files
// (region HA stage B's rule, hostfiles.PinnedFiles). Then the generation
// starts from no manifest, carrying only the env file. The claim's empty
// bootstrap reads no chain at all.
func TestEstablishHostThatNeverPublishedFiles(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	x.writeEnv("K=1\n")
	x.f.Evidence = func(context.Context, hostclaim.Observed) (hostfiles.PinnedFiles, bool, error) {
		return hostfiles.PinnedFiles{}, true, nil
	}
	o := x.observe(servingObserved("g3", "g1:uk:@root:uk", "g2:uk:g1:uk", "g3:uk:g2:uk"), "g3")
	if err := x.f.Establish(ctx, o, launchInputs("g3"), x); err != nil {
		t.Fatal(err)
	}
	m, found := x.manifest("g3")
	if !found || m.Seq != 1 || len(m.Entries) != 0 || len(m.Platform) != 1 || m.Platform[hostfiles.EnvItem].SHA256 != sha([]byte("K=1\n")) {
		t.Fatalf("never-published bootstrap: %+v %v", m, found)
	}
	// The claim's own empty bootstrap: a created incarnation before its
	// first writer, admitted after EmptyAuthority.
	x.f.Evidence = nil
	empty := x.observe(servingObserved("g1", "g1:uk:@root:uk"), "g1") // admitted: the holder at g1 after EmptyAuthority
	in := launchInputs("g1")
	in.Empty = true
	if err := x.f.Establish(ctx, empty, in, x); err != nil {
		t.Fatal(err)
	}
	if m, found := x.manifest("g1"); !found || m.Seq != 1 || len(m.Entries) != 0 || len(x.data.reads) != 0 {
		t.Fatalf("empty bootstrap: %+v %v reads %v", m, found, x.data.reads)
	}
}

// journal writes the prepared manifest's binding into the op journal
// under the fixture's data directory, as local does.
func (x *filesFixture) journal(generation string, pm preparedManifest) {
	x.t.Helper()
	x.journalRaw(generation, string(mustJSON(x.t, pm)))
}

func (x *filesFixture) journalRaw(generation, value string) {
	x.t.Helper()
	j, err := moves.ReadJournal(x.dir, filesHost, generation)
	if err != nil {
		x.t.Fatal(err)
	}
	j.Mark(landStepFilesManifest, value)
	if err := j.Commit(x.dir); err != nil {
		x.t.Fatal(err)
	}
}

// The probe's admission opens the prepared generation itself (in.Prepared,
// NewGeneration = P). Its baseline is the prepared manifest at the exact
// version the op journal's binding names (fix round 1, finding 2): those
// bytes are read by that version, checked against the journalled digest,
// validated (sequence 0, objects within the location), and must be what
// the key currently holds. The env file on disk (now carrying the minted
// key) is staged over it at sequence 1, a new version of the same key; the
// sequence-0 version stays readable by exact version and still satisfies
// stage B's rule for a prepared manifest (sequence 0, the same entries).
//
// A crash after the commit and before the local effect leaves the key at
// sequence 1: the retry recognises its own establishment (the same
// entries, the env bytes on disk, no adoption item) and writes nothing. A
// binding whose digest is not the exact version's, a current manifest that
// is neither the prepared one nor this box's establishment of it, a
// missing binding and a missing manifest all refuse before any write.
func TestEstablishOnThePreparedGenerationContinuesFromSequenceZero(t *testing.T) {
	x := newFilesFixture(t)
	ctx := context.Background()
	dir := filepath.Join(x.dir, filesHost)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil {
		t.Fatal(err)
	}
	x.writeEnv("K=1\n")
	pm, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}})
	if err != nil {
		t.Fatal(err)
	}
	if pm.Version == "" || pm.Version != x.bucket.LatestID("data", pm.Key) {
		t.Fatalf("the binding's version %q is not the manifest's exact version %q", pm.Version, x.bucket.LatestID("data", pm.Key))
	}
	x.journal("mv1", pm)
	x.writeEnv("K=1\nCARLOS_INSTANCE_KEY=abc\n")
	o := servingObserved("mv1", "cur:ie:@root:ie", "mv1:uk:cur:ie")
	o.Claim.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "mv1", Phase: "probing", Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}}
	o = x.observe(o, "mv1")
	in := launchInputs("mv1")
	in.Prepared = true
	// The first attempt crashes after the commit, before the local effect:
	// the env object and the manifest are each fenced by one With, the
	// local effect is the third, which refuses.
	x.workErr, x.workErrAt = errors.New("crash after the commit"), 3
	if err := x.f.Establish(ctx, o, in, x); err == nil {
		t.Fatal("the crash was not reported")
	}
	x.workErr, x.workErrAt = nil, 0
	m, found := x.manifest("mv1")
	if !found || m.Seq != 1 {
		t.Fatalf("after the crashed attempt: %+v %v", m, found)
	}
	seq1 := x.bucket.LatestID("data", pm.Key)
	// The retry reads its own establishment back and writes nothing.
	if err := x.f.Establish(ctx, o, in, x); err != nil {
		t.Fatalf("retry: %v", err)
	}
	m, found = x.manifest("mv1")
	if !found || m.Seq != 1 || m.Entries["a.txt"].Seq != 0 || m.Platform[hostfiles.EnvItem].SHA256 != sha([]byte("K=1\nCARLOS_INSTANCE_KEY=abc\n")) {
		t.Fatalf("after the probe's establishment: %+v %v", m, found)
	}
	if x.bucket.LatestID("data", pm.Key) != seq1 {
		t.Fatal("the retry wrote a new revision")
	}
	// The exact sequence-0 version is what the binding says, and stage B's
	// rule for a prepared manifest still holds for it: sequence 0 and the
	// same entries as the establishment.
	raw, err := x.bucket.GetVersion(ctx, "data", pm.Key, pm.Version)
	if err != nil || sha(raw) != pm.SHA256 {
		t.Fatalf("the sequence-0 version is gone or changed: %v", err)
	}
	var seq0 hostfiles.Manifest
	if err := json.Unmarshal(raw, &seq0); err != nil || seq0.Seq != 0 || !reflect.DeepEqual(seq0.Entries, m.Entries) {
		t.Fatalf("the exact prepared version: %+v %v", seq0, err)
	}
	// A binding whose digest is not the exact version's refuses.
	bad := pm
	bad.SHA256 = sha([]byte("other"))
	x.journal("mv1", bad)
	if err := x.f.Establish(ctx, o, in, x); err == nil || errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("digest mismatch: %v", err)
	}
	x.journal("mv1", pm)
	// A current manifest that is neither the prepared one nor this box's
	// establishment of it refuses, and is left as it is.
	other := m
	other.Entries = map[string]hostfiles.Entry{"b.txt": {Object: movesFilesPrefix(filesLoc, "mv1") + "objects/0000000000000002-" + sha([]byte("B")), SHA256: sha([]byte("B")), Size: 1, Seq: 2}}
	other.Seq = 2
	x.bucket.Put("data", pm.Key, mustJSON(t, other))
	if err := x.f.Establish(ctx, o, in, x); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("foreign current manifest: %v", err)
	}
	if cur, _ := x.manifest("mv1"); cur.Seq != 2 {
		t.Fatal("the foreign manifest was replaced")
	}
	// A prepared manifest naming an object outside the location (P6-D3)
	// refuses.
	foreign := hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{"x": {Object: "instances/other/files/g-z/objects/0000000000000000-" + sha(nil), SHA256: sha(nil), Seq: 0}}, Platform: map[string]hostfiles.Entry{}}
	fraw := mustJSON(t, foreign)
	fkey := hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "mv2"))
	fver := x.bucket.Put("data", fkey, fraw)
	x.journal("mv2", preparedManifest{Key: fkey, SHA256: sha(fraw), Version: fver, Op: "mv1", Stamp: 7})
	in = launchInputs("mv2")
	in.Prepared = true
	if err := x.f.Establish(ctx, o, in, x); !errors.Is(err, hostfiles.ErrOrigin) {
		t.Fatalf("foreign prepared manifest: %v", err)
	}
	// No binding in the op journal, and no manifest at P: an error, never a
	// fresh start.
	in = launchInputs("mv3")
	in.Prepared = true
	if err := x.f.Establish(ctx, o, in, x); err == nil || errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("missing binding: %v", err)
	}
	x.journal("mv3", preparedManifest{Key: hostfiles.ManifestKey(movesFilesPrefix(filesLoc, "mv3")), SHA256: sha(nil), Version: "v-none", Op: "mv1", Stamp: 7})
	if err := x.f.Establish(ctx, o, in, x); err == nil || errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("missing prepared manifest: %v", err)
	}
	if _, found := x.manifest("mv3"); found {
		t.Fatal("a manifest was written for a missing prepared generation")
	}
}