package host

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/superfly/ltx"
)

// The recovery wake owners (moves Plan 4 Task 9; ruling Q7). The fixture is
// the landing probe's (newPOFixture) turned into the source's recovery: the
// source ie-box executes the move's restoring op at probing (a rollback,
// dirty or clean) or its reverting op at probing (a revert), and the row is
// held by the rollback id or the move id.

const (
	rcDirty  = "dirty"
	rcClean  = "clean"
	rcRevert = "revert"
	// rcLegacy is a clean return of a legacy source: src_gen is the origin
	// (""), its files at the location's root and no generation.json.
	rcLegacy = "legacy"
	// rcS is the source installation's stamp S the flip (or the revert)
	// journalled.
	rcS = 15
)

// rcFiles is a fake C8 files manifest provider (Task 6 builds the real
// one): it records the call and runs under the start's work.
type rcFiles struct {
	f     *poFixture
	err   error
	calls int
}

func (r *rcFiles) Establish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	r.calls++
	r.f.order = append(r.f.order, "establish:"+o.Claim.Holder.Generation)
	if r.err != nil {
		return r.err
	}
	return work.With(ctx, in.Identity, func() error { return nil })
}

type rcFixture struct {
	*poFixture
	kind     string
	c8       *rcFiles
	inputs   *provision.MoveInputs
	manifest moves.InstallManifest // the dirty return's
	seal     moves.Seal            // the return seal
}

// rcInputs are the inputs the recovery probe recorded: neither the move's
// pins nor the landing's.
func rcInputs(env []string) *provision.MoveInputs {
	return &provision.MoveInputs{BundleETags: map[string]string{"default": "e-recovered"}, OverlayETag: "e-overlay-2", Release: "v1.3.0", EnvReport: localEnvDigest(poEnvKey, env)}
}

func newRCFixture(t *testing.T, kind string, opts ...poOption) *rcFixture {
	t.Helper()
	po := newPOFixture(t, opts...)
	f := &rcFixture{poFixture: po, kind: kind, c8: &rcFiles{f: po}}
	// The recovery runs on the source box, which is never the home box
	// here: no alias, the fleet's designation only.
	f.owners.Box, f.owners.Region = f.srcBox, f.srcRegion
	f.owners.HomeRegion, f.owners.FleetHomeRegion = "", f.home
	f.owners.Files = f.c8
	f.launch.Version = "v1.3.0"
	f.inputs = rcInputs(f.launch.Env)
	f.renderETags, f.renderOverlay = f.inputs.BundleETags, f.inputs.OverlayETag
	at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)

	// The source's own generation in its region, and the marker naming it.
	if kind == rcLegacy {
		f.bucket.Put("data", "instances/"+poHost+"/0009/"+ltx.FormatFilename(1, 30), f.ltxHeader(1, 30))
	} else {
		f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), f.ltxHeader(1, 30))
		f.bucket.Put("data", "instances/"+poHost+"/generation.json", mustJSON(t, genselect.Marker{Op: "src", Incarnation: f.inc, Host: poHost, Box: f.srcBox, InstalledAt: at}))
	}

	c := f.recoveryClaim()
	f.rec.Region = f.srcRegion
	f.rec.Move.LandingPhase = provision.LandRestored
	f.route.Status, f.route.HoldOp = "restoring", rpRB
	switch kind {
	case rcRevert:
		f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveReverting, ""
		f.rec.Move.RecoveryInputs = f.inputs
		f.route.HoldOp = poOp
	default:
		f.rec.Move.State = provision.MoveRollingBack
		f.rec.Move.Rollback = &provision.MoveRollback{ID: rpRB, Dirty: kind == rcDirty, TargetPhase: provision.RbTargetReleased, SourcePhase: provision.RbSourceProbing, RecoveryInputs: f.inputs}
	}
	if kind == rcDirty {
		// The return generation the source's install built from the return
		// lane, and its manifest in the rollback's journal (installGeneration).
		f.rec.Move.Rollback.TargetPhase = provision.RbTargetReturned
		key := "instances/" + poHost + "/g-" + rpRB + "/0009/" + ltx.FormatFilename(1, 50)
		version := f.bucket.Put("data", key, f.ltxHeader(1, 50))
		f.manifest = moves.InstallManifest{V: 1, Op: rpRB, Host: poHost, N: 50, SnapshotKey: key, SnapshotVersion: version, SnapshotCRC64: 9, DBSHA256: strings.Repeat("e", 64)}
		f.bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, rpRB)), mustJSON(t, f.manifest))
		j, err := moves.ReadJournal(f.dataDir, poHost, rpRB)
		if err != nil {
			t.Fatal(err)
		}
		body, _ := json.Marshal(f.manifest)
		j.Mark(landStepInstall, string(body))
		// The install receipt: the return's tree at the live paths.
		filesID, err := pathIdentity(f.files)
		if err != nil {
			t.Fatal(err)
		}
		envID, err := pathIdentity(filepath.Join(f.dataDir, poHost+".env"))
		if err != nil {
			t.Fatal(err)
		}
		placed, _ := json.Marshal(stagedFiles{Files: filesID, Env: envID})
		j.Mark(landStepFilesPlaced, string(placed))
		if err := j.Commit(f.dataDir); err != nil {
			t.Fatal(err)
		}
		f.seal = moves.Seal{V: moves.SealVersion, N: 50, FromDB: true, SrcCovered: true, SealedAt: at, LeaseUntil: at.Add(time.Minute),
			Replica: []moves.SealedObject{{Key: "moves/" + poHost + "/" + rpRB + "/replica/0009/" + ltx.FormatFilename(1, 50), VersionID: "r1", Size: int64(ltx.HeaderSize)}},
			Files:   []moves.SealedObject{{Key: "moves/" + poHost + "/" + rpRB + "/files/archive.sealed", VersionID: "r2", Size: 10}, {Key: "moves/" + poHost + "/" + rpRB + "/files/archive.json", VersionID: "r3", Size: 10}}}
		c.Op.Journal[moves.JournalReturnSeal] = f.seal.Encode()
		f.writeLocal(rpRB, rpRB, f.inc)
	} else {
		// The source's own files and warm database: no moves files marker.
		if err := os.Remove(filepath.Join(f.files, archive.MarkerName)); err != nil {
			t.Fatal(err)
		}
		if kind == rcLegacy {
			if err := os.Remove(f.route.DBPath + ".gen"); err != nil {
				t.Fatal(err)
			}
		} else if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", "src"); err != nil {
			t.Fatal(err)
		}
	}
	// The tree the recovery restores, as its baselines record it: a dirty
	// return's is the return baseline the target journalled, a clean
	// return's or a revert's the source's stopped baseline.
	digest, err := f.owners.filesDigest(poHost)
	if err != nil {
		t.Fatal(err)
	}
	if kind == rcDirty {
		raw, err := moves.EncodeReturnBaseline(provision.MoveSource{FinalTXID: 50, Files: &provision.MoveFilesDigest{Plain: digest, Sealed: "sealed"}, TimersSHA256: "timers", Evidence: "moves/ev"})
		if err != nil {
			t.Fatal(err)
		}
		c.Op.Journal[moves.JournalReturnBaseline] = raw
	} else {
		f.rec.Move.Source = &provision.MoveSource{FinalTXID: 30, Files: &provision.MoveFilesDigest{Plain: digest}, TimersSHA256: "timers"}
	}
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	f.putClaim(c)
	// Plan 15: the source generation the clean kinds bridge from is proven
	// only when the claim records it; record it as the #250 backfill does.
	if kind != rcLegacy {
		recordOrigins(t, f.claims, poHost, hostclaim.OriginTuple{Incarnation: f.inc, Location: "instances/" + poHost, Region: f.srcRegion, Generation: "src"})
	}
	return f
}

// recoveryClaim is the claim the recovery runs under. A return: restoring at
// probing, the source executing again at S (ReturnToSource); the holder is
// still the target's session 20 in the landing's generation, or src_gen for
// a clean return. A revert: reverting at rev_phase probing, the holder never
// left the source.
func (f *rcFixture) recoveryClaim() *hostclaim.Claim {
	h := &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + poHost + ".json", Account: "bab", App: "jam", CreatedAt: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC), Incarnation: f.inc, Location: "instances/" + poHost,
		Generation: poOp, Stamp: rcS, Session: 20, WriterBox: f.dstBox,
		Admissions: []hostclaim.AdmissionRecord{{Session: 15, Box: f.srcBox, Generation: "src", Region: f.srcRegion, Mode: "cold"},
			{Session: 20, Box: f.dstBox, Generation: poOp, Predecessor: "src", Region: f.dstEngine, PredecessorRegion: f.srcRegion, Mode: "cold", Coverage: "sealed"}}}
	journal := map[string]string{hostclaim.JournalSourceStamp: "15", hostclaim.JournalSrcGen: "src", hostclaim.JournalCutSession: "15", hostclaim.JournalEvidenceSource: "moves/bab/jam/" + poHost + "/" + poOp + "/stopped-source.json"}
	intent := map[string]string{"to": f.dstTo, "source": f.srcBox, "target": f.dstBox, "rollback": rpRB}
	if f.home != "" {
		intent["home_region"] = f.home
	}
	op := hostclaim.Op{Kind: hostclaim.KindRestoring, ID: poOp, Phase: hostclaim.PhaseProbing, Exec: f.srcBox, Participants: []string{f.srcBox, f.dstBox},
		Intent: intent, Journal: journal}
	gen := uint64(30)
	switch f.kind {
	case rcClean:
		h.Generation = "src"
	case rcLegacy:
		// The source never named a generation: its sessions predate
		// admission records, and the landing's admission came from the origin.
		h.Generation, h.Admissions = "", h.Admissions[1:]
		h.Admissions[0].Predecessor = ""
		journal[hostclaim.JournalSrcGen] = ""
	case rcRevert:
		h.Generation, h.Session, h.WriterBox, h.Admissions = "src", 15, f.srcBox, h.Admissions[:1]
		op.Kind = hostclaim.KindReverting
		delete(op.Intent, "rollback")
		journal[hostclaim.JournalRevPhase] = hostclaim.PhaseProbing
		gen = 18
	}
	return &hostclaim.Claim{V: 2, Host: poHost, Gen: gen, Holder: h, Op: op}
}

// policy is the recovery probe adapter's start policy: a probe carrying
// exactly the recorded inputs.
func (f *rcFixture) policy() activator.StartPolicy {
	in := *f.inputs
	return activator.StartPolicy{MaxStarts: 3, Window: 30 * time.Second, Probe: true, Inputs: &in}
}

func (f *rcFixture) selectAdmission(o hostclaim.Observed) ClaimLaunchInputs {
	f.t.Helper()
	in, err := f.owners.Select(f.t.Context(), f.route, o, f.policy())
	if err != nil {
		f.t.Fatal(err)
	}
	return in
}

// rcMutations are admissions Select never makes; AdmissionEvidence refuses
// every one.
var rcMutations = map[string]func(*hostclaim.Admission, *genselect.Plan){
	"stamp kept":       func(a *hostclaim.Admission, _ *genselect.Plan) { a.KeepStamp = true },
	"another region":   func(a *hostclaim.Admission, _ *genselect.Plan) { a.Region = "uk" },
	"another box":      func(a *hostclaim.Admission, _ *genselect.Plan) { a.Box = "uk-box" },
	"a stale baseline": func(a *hostclaim.Admission, p *genselect.Plan) { p.MaxTXID, a.RemoteMaxTXID = p.MaxTXID-1, p.MaxTXID-1 },
	"another session":  func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedSession = 14 },
	"warm":             func(a *hostclaim.Admission, _ *genselect.Plan) { a.Mode = "warm" },
	"a loss id":        func(a *hostclaim.Admission, _ *genselect.Plan) { a.LossID = "loss" },
	"writes lost": func(a *hostclaim.Admission, _ *genselect.Plan) {
		a.AdditionalWritesLost = []hostclaim.WritesLost{{Box: "x"}}
	},
	"certified":           func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorCertified = true },
	"another phase":       func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedPhase = hostclaim.PhaseClaimed },
	"another op":          func(a *hostclaim.Admission, _ *genselect.Plan) { a.ObservedOpID = "mv_other" },
	"the move id as gen":  func(a *hostclaim.Admission, _ *genselect.Plan) { a.NewGeneration = poOp },
	"a remote max moved":  func(a *hostclaim.Admission, _ *genselect.Plan) { a.RemoteMaxTXID++ },
	"another coverage":    func(a *hostclaim.Admission, _ *genselect.Plan) { a.Coverage = "bogus" },
	"another predecessor": func(a *hostclaim.Admission, _ *genselect.Plan) { a.PredecessorRegion = "de" },
}

func (f *rcFixture) refusesMutations(o hostclaim.Observed, in ClaimLaunchInputs) {
	f.t.Helper()
	for name, mutate := range rcMutations {
		a, p := in.Admission, *in.Baseline
		mutate(&a, &p)
		if err := f.owners.AdmissionEvidence(f.t.Context(), o, a, &p); !errors.Is(err, hostbox.ErrChanged) {
			f.t.Fatalf("%s: %v", name, err)
		}
	}
}

// Plan 4 Task 9: a dirty return's probe is a cold admission into the return
// generation (the rollback id), sealed, from the return generation's pin at
// the return seal's maximum, and it keeps no stamp: the admission moves
// holder.stamp from S to the new session and holder.generation to the
// return generation, so S is retired (a selection captured at S is refused).
func TestDirtyReturnProbeIssuesANewStamp(t *testing.T) {
	f := newRCFixture(t, rcDirty)
	ctx := t.Context()
	o := f.read()
	in := f.selectAdmission(o)
	want := hostclaim.Admission{ObservedOpID: poOp, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: 20, ObservedGeneration: poOp, Box: "ie-box", Mode: "cold",
		NewGeneration: rpRB, Coverage: "sealed", Region: "ie", PredecessorRegion: "uk", RemoteMaxTXID: 50}
	if !reflect.DeepEqual(in.Admission, want) {
		t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
	}
	if !in.Prepared || in.Baseline == nil || in.Baseline.Generation != rpRB || in.Baseline.MaxTXID != f.seal.N || in.Baseline.Pin == nil || in.Baseline.Pin.SnapshotVersion != f.manifest.SnapshotVersion {
		t.Fatalf("baseline: prepared=%v %+v", in.Prepared, in.Baseline)
	}
	if in.Identity != mpLaunchIdentity(f.inputs) || in.Version != "v1.3.0" || in.DBPath != f.route.DBPath || in.ReadyPath != "/healthz" || !contains(in.Env, "CARLOS_MOVE_PROBE=1") {
		t.Fatalf("launch inputs: %+v", in)
	}
	if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
		t.Fatal(err)
	}
	f.refusesMutations(o, in)
	unsealed := in.Admission
	unsealed.Coverage = ""
	if err := f.owners.AdmissionEvidence(ctx, o, unsealed, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("an unsealed dirty admission: %v", err)
	}

	admitted := f.admit(o, in)
	h := admitted.Claim.Holder
	last := h.Admissions[len(h.Admissions)-1]
	if h.Stamp == rcS || h.Stamp != h.Session || h.Generation != rpRB || h.WriterBox != "ie-box" || last.Coverage != "sealed" || last.WritesLost != nil || last.Predecessor != poOp {
		t.Fatalf("the admission: holder %+v, last %+v", h, last)
	}
	policy := MovesClaimPolicy(f.owners.Records)
	s := hostbox.Selection{Host: poHost, Action: movesActionSpawn, Box: "ie-box", Region: "ie", Kind: hostclaim.KindRestoring, Op: poOp, Phase: hostclaim.PhaseProbing,
		Captured: hostguard.Captured{Incarnation: f.inc, Stamp: rcS}}
	rpRefused(t, policy(ctx, admitted.Claim, s, in.Identity), "the capture is not the holder's")
	s.Captured.Stamp = h.Stamp
	if err := policy(ctx, admitted.Claim, s, in.Identity); err != nil {
		t.Fatalf("the new stamp's spawn: %v", err)
	}

	// A restarted attempt over the admitted return generation is a fresh
	// generation with no coverage, restored from the return generation's
	// pin while it has published nothing.
	again, err := f.owners.Select(ctx, f.route, admitted, f.policy())
	if err != nil {
		t.Fatal(err)
	}
	if a := again.Admission; a.NewGeneration != "fresh1" || a.KeepStamp || a.Coverage != "" || a.ObservedGeneration != rpRB || a.PredecessorRegion != "ie" || again.Prepared ||
		again.Baseline == nil || again.Baseline.Generation != rpRB || again.Baseline.MaxTXID != 50 {
		t.Fatalf("restarted: %+v prepared=%v baseline %+v", a, again.Prepared, again.Baseline)
	}
	if err := f.owners.AdmissionEvidence(ctx, admitted, again.Admission, again.Baseline); err != nil {
		t.Fatal(err)
	}
	reused := again.Admission
	reused.NewGeneration = rpRB
	if err := f.owners.AdmissionEvidence(ctx, admitted, reused, again.Baseline); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("the return generation twice: %v", err)
	}
}

// The dirty return's installation, its real admission and the real C8
// materialization together (whole-branch fixes, item 2): local built the
// prepared manifest at the source installation's stamp S (the claim's
// journalled source_stamp, under which the return's local ran) and
// journalled its binding; the recovery's cold admission issues a new
// stamp; Establish validates the binding's immutable installation stamp
// against S, separately from the admitted session's, and establishes P's
// manifest with the minted env file. A binding at the admitted session's
// stamp, or any other, is not the installation's and is refused.
func TestDirtyReturnEstablishesThePreparedManifestThroughC8(t *testing.T) {
	f := newRCFixture(t, rcDirty)
	ctx := t.Context()
	f.owners.Files = f.realC8()
	pm := f.preparedManifestJournalled(rpRB, poOp, rcS)
	o := f.read()
	in := f.selectAdmission(o)
	admitted := f.admitThroughWriter(o, in)
	h := admitted.Claim.Holder
	if h.Stamp == rcS || !in.Prepared {
		t.Fatalf("the dirty admission kept the stamp or is not prepared: %+v %v", h, in.Prepared)
	}
	if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f.poFixture}); err != nil {
		t.Fatalf("the dirty return's materialization: %v", err)
	}
	// The env file is the prepared item's (the recovery's start gate holds
	// the restored tree, env included, to the recorded digest) and there is
	// no adoption item, so the establishment reads the prepared bytes back
	// at sequence 0 and publishes nothing; what it proved is the binding.
	m, found, err := hostfiles.Load(ctx, casBucketStore{b: f.bucket, bucket: "data"}, movesFilesPrefix("instances/"+poHost, rpRB))
	if err != nil || !found || m.Seq != 0 || m.Platform[hostfiles.EnvItem].SHA256 == "" {
		t.Fatalf("P's manifest after the dirty return's probe: %+v %v %v", m, found, err)
	}
	if raw, _ := f.bucket.Current("data", pm.Key); sha(raw) != pm.SHA256 || f.bucket.LatestID("data", pm.Key) != pm.Version {
		t.Fatal("the prepared manifest is not the journalled version")
	}
	// A binding at the admitted session's stamp, or another, is refused.
	for name, stamp := range map[string]uint64{"the admitted session's stamp": h.Stamp, "another stamp": 99} {
		g := newRCFixture(t, rcDirty)
		g.owners.Files = g.realC8()
		g.preparedManifestJournalled(rpRB, poOp, stamp)
		o := g.read()
		in := g.selectAdmission(o)
		admitted := g.admitThroughWriter(o, in)
		if err := g.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: g.poFixture}); !errors.Is(err, errLandingChanged) {
			t.Fatalf("%s: %v", name, err)
		}
	}
}

// The no-mint rule (platform-1e, whole-branch fixes): a recovery probe
// preserves the restored environment byte for byte and never mints a
// CARLOS_INSTANCE_KEY; the real C8 then carries exactly those bytes as
// the generation's env item. A later change that minted during recovery
// fails here, loudly, rather than exempting <host>.env from verification.
func TestRecoveryProbeNeverMintsAnInstanceKey(t *testing.T) {
	for _, kind := range []string{rcDirty, rcClean} {
		t.Run(kind, func(t *testing.T) {
			f := newRCFixture(t, kind)
			ctx := t.Context()
			f.owners.Files = f.realC8()
			envPath := filepath.Join(f.dataDir, poHost+".env")
			restored, err := os.ReadFile(envPath)
			if err != nil {
				t.Fatal(err)
			}
			if strings.Contains(string(restored), activator.InstanceKeyVar+"=") {
				t.Fatalf("the restored env already carries a key: %q", restored)
			}
			gen := "src"
			if kind == rcDirty {
				gen = rpRB
			}
			// The manifest the recovery establishes from: the source's, or the
			// dirty return's prepared one, built from the restored tree.
			pm := f.preparedManifestJournalled(gen, poOp, rcS)
			o := f.read()
			in := f.selectAdmission(o)
			admitted := f.admitThroughWriter(o, in)
			if kind == rcClean {
				// The restore placed the admitted generation's database (the
				// start check's, not C8's).
				if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", admitted.Claim.Holder.Generation); err != nil {
					t.Fatal(err)
				}
			}
			if err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f.poFixture}); err != nil {
				t.Fatalf("the recovery's materialization: %v", err)
			}
			after, err := os.ReadFile(envPath)
			if err != nil {
				t.Fatal(err)
			}
			if string(after) != string(restored) {
				t.Fatalf("the recovery changed the restored env:\n%q\n%q", restored, after)
			}
			if strings.Contains(string(after), activator.InstanceKeyVar+"=") {
				t.Fatalf("the recovery minted a key: %q", after)
			}
			m, found, err := hostfiles.Load(ctx, casBucketStore{b: f.bucket, bucket: "data"}, movesFilesPrefix("instances/"+poHost, admitted.Claim.Holder.Generation))
			if err != nil || !found || m.Platform[hostfiles.EnvItem].SHA256 != sha(restored) {
				t.Fatalf("the env item is not the restored env: %+v %v %v (prepared %s)", m.Platform[hostfiles.EnvItem], found, err, pm.Key)
			}
		})
	}
}

// Plan 4 Task 9: a clean return's probe is a cold admission into a fresh
// generation restored from src_gen in this region, its coverage
// clean-return: the admission records no writes lost.
func TestCleanReturnProbeIsCoveredCleanReturn(t *testing.T) {
	f := newRCFixture(t, rcClean)
	ctx := t.Context()
	o := f.read()
	in := f.selectAdmission(o)
	want := hostclaim.Admission{ObservedOpID: poOp, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: 20, ObservedGeneration: "src", Box: "ie-box", Mode: "cold",
		NewGeneration: "fresh1", Coverage: "clean-return", Region: "ie", PredecessorRegion: "uk", RemoteMaxTXID: 30}
	if !reflect.DeepEqual(in.Admission, want) {
		t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
	}
	if in.Prepared || in.Baseline == nil || in.Baseline.Generation != "src" || in.Baseline.MaxTXID != 30 {
		t.Fatalf("baseline: prepared=%v %+v", in.Prepared, in.Baseline)
	}
	if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
		t.Fatal(err)
	}
	f.refusesMutations(o, in)
	for _, coverage := range []string{"", "sealed"} {
		a := in.Admission
		a.Coverage = coverage
		if err := f.owners.AdmissionEvidence(ctx, o, a, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("coverage %q: %v", coverage, err)
		}
	}
	admitted := f.admit(o, in)
	h := admitted.Claim.Holder
	last := h.Admissions[len(h.Admissions)-1]
	if last.Coverage != "clean-return" || last.WritesLost != nil || len(last.AdditionalWritesLost) != 0 || h.Generation != "fresh1" || h.Stamp == rcS {
		t.Fatalf("the admission: holder %+v, last %+v", h, last)
	}

	// A clean return whose holder does not name src_gen is not this return.
	t.Run("the holder is not at src_gen", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		c := f.recoveryClaim()
		c.Holder.Generation = poOp
		f.putClaim(c)
		if _, err := f.owners.Select(t.Context(), f.route, f.read(), f.policy()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
	})
	// Fix round 1, ruling (4): AdmitWriter records the target's last
	// admission as the recovery admission's predecessor, so with that
	// admission covered the ordinary selection of a restart that published
	// no root ends in ErrCovered. Until the recovery has a root, a clean
	// return's restart restores the recorded source baseline (src_gen, by
	// exact version); after its first root ordinary selection resumes.
	restart := func(t *testing.T) (*rcFixture, hostclaim.Observed, ClaimLaunchInputs) {
		f := newRCFixture(t, rcClean)
		c := f.recoveryClaim()
		c.Holder.Admissions[1].Covered = true
		f.putClaim(c)
		o := f.read()
		in := f.selectAdmission(o)
		admitted := f.admitThroughWriter(o, in)
		if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", "fresh1"); err != nil {
			t.Fatal(err)
		}
		if err := f.owners.Materialize(t.Context(), admitted, in, poWork{action: "spawn", f: f.poFixture}); err != nil {
			t.Fatal(err)
		}
		return f, admitted, in
	}
	t.Run("a restart before the first root restores src_gen", func(t *testing.T) {
		f, admitted, first := restart(t)
		ctx := t.Context()
		again, err := f.owners.Select(ctx, f.route, admitted, f.policy())
		if err != nil {
			t.Fatal(err)
		}
		if a := again.Admission; a.NewGeneration != "fresh2" || a.Coverage != "" || a.ObservedGeneration != "fresh1" || a.RemoteMaxTXID != 30 || again.Prepared ||
			again.Baseline.Generation != "src" || !reflect.DeepEqual(again.Baseline.Files, first.Baseline.Files) {
			t.Fatalf("restart: %+v baseline %+v", a, again.Baseline)
		}
		if err := f.owners.AdmissionEvidence(ctx, admitted, again.Admission, again.Baseline); err != nil {
			t.Fatal(err)
		}
		restarted := f.admit(admitted, again)
		// The recovery's first root: ordinary selection resumes.
		f.bucket.Put("data", "instances/"+poHost+"/g-fresh2/0009/"+ltx.FormatFilename(1, 31), f.ltxHeader(1, 31))
		third, err := f.owners.Select(ctx, f.route, restarted, f.policy())
		if err != nil {
			t.Fatal(err)
		}
		if third.Baseline.Generation != "fresh2" || third.Admission.RemoteMaxTXID != 31 || third.Admission.NewGeneration != "fresh3" {
			t.Fatalf("after the first root: %+v baseline %+v", third.Admission, third.Baseline)
		}
		if err := f.owners.AdmissionEvidence(ctx, restarted, third.Admission, third.Baseline); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("the source baseline is the recorded one, by exact version", func(t *testing.T) {
		f, admitted, _ := restart(t)
		f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), f.ltxHeader(1, 30))
		if _, err := f.owners.Select(t.Context(), f.route, admitted, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("another version of src_gen: %v", err)
		}
	})
	// Fix round 2: the source baseline is persisted before the restore (the
	// activator restores before Materialize), at the admission's evidence,
	// so a crash between the restore and Materialize still pins it.
	t.Run("a crash between the restore and Materialize keeps the pin", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		c := f.recoveryClaim()
		c.Holder.Admissions[1].Covered = true
		f.putClaim(c)
		ctx := t.Context()
		o := f.read()
		in := f.selectAdmission(o)
		admitted := f.admitThroughWriter(o, in)
		// The activator restores in.Baseline here; the box crashes before
		// Materialize. src_gen then shows another version.
		f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), f.ltxHeader(1, 30))
		if _, err := f.owners.Select(ctx, f.route, admitted, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("the restart after the crash: %v", err)
		}
	})
	// Fix round 2: a legacy source's src_gen is the origin, "" in the claim
	// and hostclaim.RootGeneration in a selection; the source baseline and
	// its pin compare the two as one.
	t.Run("a legacy source's clean return", func(t *testing.T) {
		f := newRCFixture(t, rcLegacy)
		ctx := t.Context()
		o := f.read()
		in := f.selectAdmission(o)
		if a := in.Admission; a.NewGeneration != "fresh1" || a.Coverage != "clean-return" || a.ObservedGeneration != "" || a.RemoteMaxTXID != 30 ||
			in.Baseline.Generation != hostclaim.RootGeneration {
			t.Fatalf("admission %+v baseline %+v", a, in.Baseline)
		}
		admitted := f.admitThroughWriter(o, in)
		again, err := f.owners.Select(ctx, f.route, admitted, f.policy())
		if err != nil {
			t.Fatal(err)
		}
		if again.Baseline.Generation != hostclaim.RootGeneration || again.Admission.RemoteMaxTXID != 30 || again.Admission.Coverage != "" {
			t.Fatalf("restart %+v baseline %+v", again.Admission, again.Baseline)
		}
		if err := f.owners.AdmissionEvidence(ctx, admitted, again.Admission, again.Baseline); err != nil {
			t.Fatal(err)
		}
		// The origin's pin was journalled at the first admission.
		f.bucket.Put("data", "instances/"+poHost+"/0009/"+ltx.FormatFilename(1, 30), f.ltxHeader(1, 30))
		if _, err := f.owners.Select(ctx, f.route, admitted, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("another version of the origin: %v", err)
		}
	})
	// A declared or fenced return (Q6) asks no coverage: the claim journals
	// the fence and the record carries it as its loss.
	t.Run("a fenced return asks no coverage", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		c := f.recoveryClaim()
		c.Op.Journal["fence.uk-box"] = `{"key":"fences/uk-box"}`
		f.putClaim(c)
		f.rec.Move.Rollback.LossRef = "fences/uk-box"
		o := f.read()
		in := f.selectAdmission(o)
		if in.Admission.Coverage != "" || in.Admission.NewGeneration != "fresh1" {
			t.Fatalf("admission %+v", in.Admission)
		}
		if err := f.owners.AdmissionEvidence(t.Context(), o, in.Admission, in.Baseline); err != nil {
			t.Fatal(err)
		}
		covered := in.Admission
		covered.Coverage = "clean-return"
		if err := f.owners.AdmissionEvidence(t.Context(), o, covered, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
		// The record and the claim disagree: wait.
		f.rec.Move.Rollback.LossRef = ""
		if _, err := f.owners.Select(t.Context(), f.route, o, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
	})
}

// A revert's probe (ruling Q7): a cold admission into a fresh generation
// restored from the holder's own generation, which a revert never changed,
// with no coverage: the admission records the source's writes as possibly
// lost (uncovered), and keeps no stamp.
func TestRevertProbeAdmitsUncovered(t *testing.T) {
	f := newRCFixture(t, rcRevert)
	ctx := t.Context()
	o := f.read()
	in := f.selectAdmission(o)
	want := hostclaim.Admission{ObservedOpID: poOp, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: 15, ObservedGeneration: "src", Box: "ie-box", Mode: "cold",
		NewGeneration: "fresh1", Region: "ie", PredecessorRegion: "ie", RemoteMaxTXID: 30}
	if !reflect.DeepEqual(in.Admission, want) {
		t.Fatalf("admission:\n got %+v\nwant %+v", in.Admission, want)
	}
	if in.Prepared || in.Baseline == nil || in.Baseline.Generation != "src" || in.Identity != mpLaunchIdentity(f.inputs) {
		t.Fatalf("inputs: %+v", in)
	}
	if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
		t.Fatal(err)
	}
	f.refusesMutations(o, in)
	for _, coverage := range []string{"clean-return", "sealed"} {
		a := in.Admission
		a.Coverage = coverage
		if err := f.owners.AdmissionEvidence(ctx, o, a, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("coverage %q: %v", coverage, err)
		}
	}
	admitted := f.admit(o, in)
	h := admitted.Claim.Holder
	last := h.Admissions[len(h.Admissions)-1]
	if last.Coverage != "" || last.WritesLost == nil || last.WritesLost.Box != "ie-box" || last.WritesLost.Session != 15 || last.WritesLost.Generation != "src" || last.WritesLost.RemoteMaxTXID != 30 ||
		h.Generation != "fresh1" || h.Stamp == rcS {
		t.Fatalf("the admission: holder %+v, last %+v", h, last)
	}

	// The revert's inputs are exactly the recorded ones, and its row is held
	// by the move id.
	for name, mutate := range map[string]func(*rcFixture, *activator.StartPolicy){
		"other inputs":         func(_ *rcFixture, p *activator.StartPolicy) { p.Inputs.Release = "v9" },
		"no inputs":            func(_ *rcFixture, p *activator.StartPolicy) { p.Inputs = nil },
		"nothing recorded":     func(f *rcFixture, _ *activator.StartPolicy) { f.rec.Move.RecoveryInputs = nil },
		"not a probe":          func(_ *rcFixture, p *activator.StartPolicy) { p.Probe = false },
		"a rollback's hold":    func(f *rcFixture, _ *activator.StartPolicy) { f.route.HoldOp = rpRB },
		"reopened":             func(f *rcFixture, _ *activator.StartPolicy) { now := time.Now(); f.rec.Move.ReopenedAt = &now },
		"another live release": func(f *rcFixture, _ *activator.StartPolicy) { f.launch.Version = "v1.2.3" },
	} {
		t.Run(name, func(t *testing.T) {
			f := newRCFixture(t, rcRevert)
			p := f.policy()
			mutate(f, &p)
			if _, err := f.owners.Select(t.Context(), f.route, f.read(), p); err == nil {
				t.Fatal("selected")
			}
		})
	}
}

// Ruling Q7: the moves owner set dispatches by the observed op's kind. A
// landing reaches the landing probe's owners, a restoring or reverting op
// the recovery owners, and the moving kind nothing; without C8 the
// recovery kinds refuse, typed, while the landing is unchanged.
func TestMovesOwnersDispatchByKind(t *testing.T) {
	ctx := t.Context()
	dispatch := func(m *movesProbeOwners) ClaimOwnerDispatch {
		owners := m.MovesOwners(&movesProbeAttempts{}).ClaimKindOwners
		return ClaimOwnerDispatch{Moves: &owners}
	}

	landing := newPOFixture(t)
	landing.owners.Files = &rcFiles{f: landing}
	in, err := dispatch(landing.owners).Select(ctx, landing.route, landing.read(), poProbe)
	if err != nil || in.Admission.NewGeneration != poOp || in.Admission.Region != "uk" {
		t.Fatalf("landing: %+v %v", in.Admission, err)
	}

	for _, kind := range []string{rcDirty, rcClean, rcRevert} {
		f := newRCFixture(t, kind)
		d := dispatch(f.owners)
		o := f.read()
		in, err := d.Select(ctx, f.route, o, f.policy())
		if err != nil {
			t.Fatalf("%s: %v", kind, err)
		}
		if in.Admission.Region != "ie" || in.Admission.Box != "ie-box" || (kind == rcDirty) != (in.Admission.NewGeneration == rpRB) {
			t.Fatalf("%s: %+v", kind, in.Admission)
		}
		if err := d.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
			t.Fatalf("%s evidence: %v", kind, err)
		}
		if err := d.EmptyAuthority(ctx, o, in); !errors.Is(err, hostbox.ErrUnsupported) {
			t.Fatalf("%s empty authority: %v", kind, err)
		}

		// Without C8 the recovery refuses before reading anything.
		f.owners.Files = nil
		f.launches = 0
		_, err = d.Select(ctx, f.route, o, f.policy())
		var missing *ClaimDependencyError
		if !errors.As(err, &missing) || !strings.Contains(missing.Error(), "C8") || f.launches != 0 {
			t.Fatalf("%s without C8: %v", kind, err)
		}
	}
	landing.owners.Files = nil
	if _, err := dispatch(landing.owners).Select(ctx, landing.route, landing.read(), poProbe); err != nil {
		t.Fatalf("the landing without C8: %v", err)
	}

	// The moving kind reaches no owner.
	moving := newPOFixture(t)
	c := moving.poFlipClaim()
	c.Op.Kind, c.Op.Phase = hostclaim.KindMoving, hostclaim.PhaseSealed
	moving.putClaim(c)
	if _, err := dispatch(moving.owners).Select(ctx, moving.route, moving.read(), poProbe); !errors.Is(err, hostbox.ErrUnsupported) {
		t.Fatalf("moving: %v", err)
	}
}

// Ruling Q7's Materialize: before every start of the admitted recovery
// probe, the held render and the held release of the recorded inputs, each
// under its own held work at the admitted capture, then the return
// generation's verified marker (dirty), then C8's Establish under the
// start's work, then the start check. A changed render, a refused held
// work or a refused Establish stops the start.
func TestRecoveryMaterializeRendersTheRecordedInputsUnderHeldWork(t *testing.T) {
	ctx := t.Context()
	start := func(t *testing.T, kind string) (*rcFixture, hostclaim.Observed, ClaimLaunchInputs) {
		f := newRCFixture(t, kind)
		o := f.read()
		in := f.selectAdmission(o)
		admitted := f.admitThroughWriter(o, in)
		if kind != rcDirty {
			// The restore rewrites <dbPath>.gen to the admitted generation.
			if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", admitted.Claim.Holder.Generation); err != nil {
				t.Fatal(err)
			}
		}
		f.order = nil
		return f, admitted, in
	}
	cfg := func(f *rcFixture) string {
		return HeldConfigIdentity(f.inputs.BundleETags, f.inputs.OverlayETag).Config
	}
	launch := func(f *rcFixture) string { id := mpLaunchIdentity(f.inputs); return id.Config + "|" + id.Release }

	t.Run("dirty", func(t *testing.T) {
		f, o, in := start(t, rcDirty)
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); err != nil {
			t.Fatal(err)
		}
		// The admitted writer's fence first, then the held render and release
		// each under its own selection, then C8 under the start's work.
		rcInOrder(t, f.order, "spawn:"+launch(f), "render", "select:held-config", "held-config:"+cfg(f)+"|", "select:held-release", "release:v1.3.0", "held-release:|v1.3.0",
			"establish:"+rpRB, "spawn:"+launch(f))
		if f.c8.calls != 1 {
			t.Fatalf("C8 ran %d times", f.c8.calls)
		}
		// The verified marker names the return generation with its pin.
		marker, err := f.owners.readMarker(ctx, o.Claim.Holder)
		if err != nil || marker == nil || marker.Op != rpRB || marker.Pin != manifestPin(f.manifest) || marker.Box != "ie-box" {
			t.Fatalf("marker %+v %v", marker, err)
		}
		if err := f.owners.Publish(ctx, o, in); err != nil {
			t.Fatal(err)
		}
		if err := f.owners.PublicationEvidence(ctx, o); err != nil {
			t.Fatalf("the return generation's publication: %v", err)
		}
	})
	t.Run("clean", func(t *testing.T) {
		f, o, in := start(t, rcClean)
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); err != nil {
			t.Fatal(err)
		}
		rcInOrder(t, f.order, "render", "held-config:"+cfg(f)+"|", "release:v1.3.0", "held-release:|v1.3.0", "establish:fresh1")
		if f.c8.calls != 1 {
			t.Fatalf("C8 ran %d times", f.c8.calls)
		}
		// No prepared generation: the marker still names src_gen until the
		// fenced publication.
		if marker, err := f.owners.readMarker(ctx, o.Claim.Holder); err != nil || marker.Op != "src" {
			t.Fatalf("marker %+v %v", marker, err)
		}
		if err := f.owners.Publish(ctx, o, in); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("an unfenced publication of a fresh generation: %v", err)
		}
		// The fenced publication after the session's own first root.
		f.bucket.Put("data", "instances/"+poHost+"/g-fresh1/0009/"+ltx.FormatFilename(1, 31), f.ltxHeader(1, 31))
		checks := 0
		pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { checks++; return nil }}
		receipt, err := f.owners.PublishFenced(ctx, o, in, pub)
		if err != nil || receipt.MarkerVersion == "" || checks == 0 {
			t.Fatalf("PublishFenced: %+v %v (checks %d)", receipt, err, checks)
		}
		if marker, err := f.owners.readMarker(ctx, o.Claim.Holder); err != nil || marker.Op != "fresh1" || marker.Box != "ie-box" {
			t.Fatalf("marker %+v %v", marker, err)
		}
		if err := f.owners.PublicationEvidence(ctx, o); err != nil {
			t.Fatalf("the fresh generation's publication: %v", err)
		}
	})
	t.Run("the render is not the recorded inputs", func(t *testing.T) {
		f, o, in := start(t, rcRevert)
		f.renderETags = map[string]string{"default": "e-moved-on"}
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
		if contains(f.order, "select:held-config") || f.c8.calls != 0 {
			t.Fatalf("selected or established after a changed render: %q", f.order)
		}
	})
	t.Run("the launch environment is not the recorded one", func(t *testing.T) {
		f, o, in := start(t, rcClean)
		in.Env = append(in.Env, "B=2")
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
		if f.c8.calls != 0 {
			t.Fatal("established")
		}
	})
	t.Run("another identity", func(t *testing.T) {
		f, o, in := start(t, rcClean)
		in.Identity = mpLaunchIdentity(mpPins())
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
		if len(f.order) != 0 {
			t.Fatalf("ran %q", f.order)
		}
	})
	t.Run("the start's work is refused", func(t *testing.T) {
		f, o, in := start(t, rcDirty)
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture, shut: true}); !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatal(err)
		}
		if contains(f.order, "render") || f.c8.calls != 0 {
			t.Fatalf("ran %q", f.order)
		}
	})
	t.Run("C8 refuses", func(t *testing.T) {
		f, o, in := start(t, rcClean)
		f.c8.err = errors.New("the bucket cannot be reached")
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); !errors.Is(err, f.c8.err) {
			t.Fatal(err)
		}
	})
	t.Run("not the admitted session", func(t *testing.T) {
		f := newRCFixture(t, rcClean)
		o := f.read()
		in := f.selectAdmission(o)
		if err := f.owners.Materialize(ctx, o, in, poWork{action: "spawn", f: f.poFixture}); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatal(err)
		}
		if len(f.order) != 0 {
			t.Fatalf("ran %q", f.order)
		}
	})
}

// rcInOrder requires each of want in order, as a subsequence of got.
func rcInOrder(t *testing.T, got []string, want ...string) {
	t.Helper()
	i := 0
	for _, g := range got {
		if i < len(want) && g == want[i] {
			i++
		}
	}
	if i != len(want) {
		t.Fatalf("order:\n got %q\nwant, in order, %q (missing from %q)", got, want, want[i])
	}
}

// Fix round 3 (ruling 3, Astra's rule): the placed tree's journalled
// digest is verified only at the start gate, before the recovery's first
// authorized start. From then on the tree is the recovery's own: ownership
// rests on its admitted process lineage under the claim session's fence,
// and its writes are authoritative (spec §7.8). The relabel checks the
// fence, the placement identity (a replaced tree) and the label's
// predecessor and incarnation, never the content. A later attempt over the
// same tree is admitted through the same lineage, so an earlier attempt's
// writes never block it; an admission the lineage cannot prove is refused.
func TestRecoveryPublicationRelabelsTheRecoverysOwnTree(t *testing.T) {
	changeAFile := func(t *testing.T, f *rcFixture) {
		if err := os.WriteFile(filepath.Join(f.files, "a.txt"), []byte("written by the recovery's app"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	replace := func(t *testing.T, f *rcFixture) {
		label, err := os.ReadFile(filepath.Join(f.files, archive.MarkerName))
		if err != nil {
			t.Fatal(err)
		}
		if err := os.Rename(f.files, f.files+"-replaced"); err != nil {
			t.Fatal(err)
		}
		if err := os.Mkdir(f.files, 0o700); err != nil {
			t.Fatal(err)
		}
		for name, body := range map[string][]byte{"a.txt": []byte("the landing's file"), archive.MarkerName: label} {
			if err := os.WriteFile(filepath.Join(f.files, name), body, 0o600); err != nil {
				t.Fatal(err)
			}
		}
	}
	unproven := func(t *testing.T, f *rcFixture) {
		err := moves.UpdateJournal(f.dataDir, poHost, rpRB, func(j *moves.Journal) error {
			j.Unmark(rcStepLineage + "fresh1")
			return nil
		})
		if err != nil {
			t.Fatal(err)
		}
	}
	for _, tc := range []struct {
		name                    string
		earlier, before, during func(t *testing.T, f *rcFixture)
		ok                      bool
	}{
		{name: "the tree as placed", ok: true},
		{name: "a restart after an earlier attempt wrote files", earlier: changeAFile, ok: true},
		{name: "authorized writes before the publication", before: changeAFile, ok: true},
		{name: "authorized writes during the publication", during: changeAFile, ok: true},
		{name: "the tree is replaced before the publication", before: replace},
		{name: "the tree is replaced during the publication", during: replace},
		{name: "an admission the lineage cannot prove", before: unproven},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newRCFixture(t, rcDirty)
			ctx := t.Context()
			work := poWork{action: "spawn", f: f.poFixture}
			o := f.read()
			in := f.selectAdmission(o)
			admitted := f.admitThroughWriter(o, in)
			if err := f.owners.Materialize(ctx, admitted, in, work); err != nil {
				t.Fatal(err)
			}
			if err := f.owners.Publish(ctx, admitted, in); err != nil {
				t.Fatal(err)
			}
			if tc.earlier != nil {
				tc.earlier(t, f)
			}
			// A restart over the published return generation.
			again, err := f.owners.Select(ctx, f.route, admitted, f.policy())
			if err != nil {
				t.Fatal(err)
			}
			restarted := f.admitThroughWriter(admitted, again)
			if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", "fresh1"); err != nil {
				t.Fatal(err)
			}
			if err := f.owners.Materialize(ctx, restarted, again, work); err != nil {
				t.Fatal(err)
			}
			f.bucket.Put("data", "instances/"+poHost+"/g-fresh1/0009/"+ltx.FormatFilename(1, 51), f.ltxHeader(1, 51))
			if tc.before != nil {
				tc.before(t, f)
			}
			pub := ClaimPublication{Check: func(context.Context) error { return nil }, Sync: func(context.Context) error {
				if tc.during != nil {
					tc.during(t, f)
				}
				return nil
			}}
			_, err = f.owners.PublishFenced(ctx, restarted, again, pub)
			raw, rerr := os.ReadFile(filepath.Join(f.files, archive.MarkerName))
			if rerr != nil {
				t.Fatal(rerr)
			}
			var label moves.FilesGen
			if err := json.Unmarshal(raw, &label); err != nil {
				t.Fatal(err)
			}
			if tc.ok {
				if err != nil || label.Op != "fresh1" {
					t.Fatalf("publication %v, label %+v", err, label)
				}
				return
			}
			if !errors.Is(err, hostbox.ErrChanged) || label.Op != rpRB {
				t.Fatalf("publication %v, label %+v: want refused and still %s", err, label, rpRB)
			}
		})
	}
}

// Fix round 3: before the recovery's first authorized start, the start gate
// requires the tree to be the one the recovery restored: a dirty return's
// the return baseline's files digest, a clean return's or a revert's the
// source's stopped baseline. A file changed before the first start is
// refused, nothing is established, and no placement is journalled, so the
// next attempt checks again.
func TestRecoveryStartGateRefusesATreeChangedBeforeTheFirstStart(t *testing.T) {
	for _, kind := range []string{rcDirty, rcClean, rcRevert} {
		t.Run(kind, func(t *testing.T) {
			f := newRCFixture(t, kind)
			ctx := t.Context()
			o := f.read()
			in := f.selectAdmission(o)
			admitted := f.admitThroughWriter(o, in)
			if kind != rcDirty {
				if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", admitted.Claim.Holder.Generation); err != nil {
					t.Fatal(err)
				}
			}
			if err := os.WriteFile(filepath.Join(f.files, "a.txt"), []byte("tampered before the first start"), 0o600); err != nil {
				t.Fatal(err)
			}
			err := f.owners.Materialize(ctx, admitted, in, poWork{action: "spawn", f: f.poFixture})
			if !errors.Is(err, hostbox.ErrChanged) || f.c8.calls != 0 {
				t.Fatalf("materialize %v, C8 ran %d times", err, f.c8.calls)
			}
			j, jerr := moves.ReadJournal(f.dataDir, poHost, map[string]string{rcDirty: rpRB, rcClean: rpRB, rcRevert: poOp}[kind])
			if jerr != nil {
				t.Fatal(jerr)
			}
			if _, ok := j.Steps[rcStepFilesPlaced]; ok {
				t.Fatal("a refused tree was journalled as placed")
			}
		})
	}
}

// admitThroughWriter admits through the claim's writer, as the wake does:
// the owners' AdmissionEvidence, then the CAS.
func (f *rcFixture) admitThroughWriter(o hostclaim.Observed, in ClaimLaunchInputs) hostclaim.Observed {
	f.t.Helper()
	w := hostbox.Writer{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: f.srcRegion, Box: f.srcBox, AdmissionEvidence: f.owners.AdmissionEvidence}
	admitted, err := w.Admit(f.t.Context(), o, in.Admission, in.Baseline)
	if err != nil {
		f.t.Fatal(err)
	}
	return admitted
}

// Fix round 4: the invariant lineage relies on (see its comment). While the
// op is restoring or reverting, the claim admits no writer outside the
// recovery owners: the writer refuses any warm admission (the reviewer's
// counterexample keeps G under an outside session); only the executor may
// admit; and the dispatch hands every cold admission of the op to the
// recovery owners' evidence, which refuses one they would not select (an
// outside owner's shape) and journals the link of any it accepts, so even a
// hand-built admission that passes is in the recovery's lineage.
func TestRecoveryOpAdmitsNoOutsideWriter(t *testing.T) {
	for _, kind := range []string{rcClean, rcRevert} {
		t.Run(kind, func(t *testing.T) {
			f := newRCFixture(t, kind)
			ctx := t.Context()
			o := f.read()
			in := f.selectAdmission(o)
			admitted := f.admitThroughWriter(o, in)
			outside := 0
			serving := ClaimKindOwners{Inputs: f.owners, PublicationEvidence: func(context.Context, hostclaim.Observed) error { return nil },
				AdmissionEvidence: func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error {
					outside++
					return nil
				}}
			moves := f.owners.MovesOwners(&movesProbeAttempts{}).ClaimKindOwners
			dispatch := ClaimOwnerDispatch{Serving: serving, Moves: &moves}
			w := hostbox.Writer{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: "ie", Box: "ie-box", AdmissionEvidence: dispatch.AdmissionEvidence}
			h := admitted.Claim.Holder
			plan, err := genselect.Select(ctx, f.bucket, f.owners.request(poHost, h))
			if err != nil {
				t.Fatal(err)
			}
			restart := hostclaim.Admission{ObservedOpID: poOp, ObservedPhase: hostclaim.PhaseProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation,
				Box: "ie-box", Region: "ie", PredecessorRegion: "ie", Mode: "cold", NewGeneration: "outside1", RemoteMaxTXID: plan.MaxTXID}
			unmoved := func() {
				t.Helper()
				now := f.read().Claim.Holder
				if now.Session != h.Session || now.Generation != h.Generation || now.WriterBox != h.WriterBox || len(now.Admissions) != len(h.Admissions) {
					t.Fatalf("the holder moved: %+v", now)
				}
			}

			// The counterexample's outside warm admission keeping G.
			warm := restart
			warm.Mode, warm.NewGeneration, warm.PredecessorCertified = "warm", "", true
			if _, err := w.Admit(ctx, admitted, warm, nil); !errors.Is(err, hostbox.ErrUnsupported) {
				t.Fatalf("a warm admission: %v", err)
			}
			// An outside owner's shape (serving's: no op, no phase), and one
			// claiming coverage the recovery's restart never claims.
			for name, mutate := range map[string]func(*hostclaim.Admission){
				"serving's shape":  func(a *hostclaim.Admission) { a.ObservedOpID, a.ObservedPhase = "", "" },
				"claimed coverage": func(a *hostclaim.Admission) { a.Coverage = "clean-return" },
				"a kept stamp":     func(a *hostclaim.Admission) { a.KeepStamp = true },
			} {
				a := restart
				mutate(&a)
				if _, err := w.Admit(ctx, admitted, a, plan); err == nil {
					t.Fatalf("%s was admitted", name)
				}
			}
			if outside != 0 {
				t.Fatalf("the dispatch reached another owner's evidence %d times", outside)
			}
			// Another box is never the executor.
			other := restart
			other.Box = "uk-box"
			if _, err := hostclaim.Update(ctx, f.claims, admitted, hostclaim.AdmitWriter(other)); !errors.Is(err, hostclaim.ErrWrongState) {
				t.Fatalf("another box: %v", err)
			}
			unmoved()

			// Any cold admission the op does take passed the recovery's
			// evidence, which journalled its link: it is in the lineage.
			taken, err := w.Admit(ctx, admitted, restart, plan)
			if err != nil {
				t.Fatal(err)
			}
			v, err := f.owners.recovery().recovery(ctx, taken)
			if err != nil {
				t.Fatal(err)
			}
			if err := f.owners.recovery().lineage(poHost, v); err != nil {
				t.Fatalf("an admission the evidence passed: %v", err)
			}
		})
	}
}