package host

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

func stagedKeys(h *phase1Harness, prefix string) []string {
	var keys []string
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			keys = append(keys, strings.TrimPrefix(k, prefix))
		}
	}
	sort.Strings(keys)
	return keys
}

// The whole preparing round on the new layout: the replica lane holds the
// claim's generation flat under replica/<level>/, move.json is there, the
// archive is staged, and the move is not promoted until a passing, fresh
// target report exists; the target's own pass writes one, and the next
// source round promotes, pinning the inputs the target checked.
func TestPhase1PrepareAndPromote(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")

	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
		t.Fatalf("after the first round: %s, want preparing", got)
	}
	lane := moves.StagingPrefix(p1Host, p1ID)
	keys := stagedKeys(h, lane)
	want := []string{"move.json", "replica/0000/0000000000000001-0000000000000002.ltx", "replica/0001/0000000000000001-0000000000000002.ltx"}
	for _, k := range want {
		if !containsString(keys, k) {
			t.Fatalf("the lane lacks %s: %v", k, keys)
		}
	}
	for _, k := range keys {
		if strings.HasPrefix(k, "replica/") && !containsString(want, k) {
			t.Fatalf("the lane holds %s, outside the claim's generation", k)
		}
	}
	sr := h.sourceReport(t)
	if sr == nil || sr.Archive == nil || sr.FilesPlain == "" || sr.Release != "v7" || !sr.S1.OK || !sr.S2.OK || sr.Env.Digest == "" || sr.CrossRate <= 0 {
		t.Fatalf("source report: %+v", sr)
	}
	// No target report: still preparing, however many rounds.
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
		t.Fatalf("promoted without a target report: %s", got)
	}

	h.targetPass(t, h.targetCfg())
	tr := h.targetReport(t)
	if ok, why := tr.Passing(); !ok {
		t.Fatalf("the target's report does not pass: %s (%+v)", why, tr.Checks)
	}
	h.pass(t)
	got := h.read(t, p1Host)
	if got.Move.State != provision.MoveReady || got.Move.Inputs == nil || got.Move.Inputs.FilesPlain != sr.FilesPlain || got.Move.Inputs.Release != "v7" || got.Move.Inputs.EnvReport != sr.Env.Digest {
		t.Fatalf("after a passing target report: %+v inputs %+v", got.Move, got.Move.Inputs)
	}
}

// A stale target report after ready demotes nothing (Review Focus 5): the
// worker keeps mirroring a ready move and never writes it back.
func TestPhase1StaleReportAfterReadyDemotesNothing(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MoveReady {
		t.Fatalf("not ready: %s", got)
	}
	h.clock.Advance(2 * moves.ReportFreshness)
	h.pass(t)
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MoveReady {
		t.Fatalf("a stale report demoted a ready move to %s", got)
	}
}

// With the switch off, an id'd record is inert: no lease, no staging write,
// no claim write, no CAS, and one log line however many passes run. The
// exec backend's Strict stays nil and Phase1 false, so every spawn and
// restore is main's.
func TestPhase1IsInertWithTheSwitchOff(t *testing.T) {
	h := newPhase1Harness(t)
	before := h.start(t, "auto")
	claimBefore := h.claim(t)
	h.cfg.Enabled = false
	logs := captureLog(t)
	h.pass(t)
	h.pass(t)
	if n := len(h.f.stagingCalls()); n != 0 || h.lease.acquired != 0 {
		t.Fatalf("switch off: %d staging requests, %d leases", n, h.lease.acquired)
	}
	for _, op := range h.f.ops {
		if strings.HasPrefix(op, "copy") || strings.Contains(op, "moves/") {
			t.Fatalf("switch off: %s", op)
		}
	}
	if got := h.read(t, p1Host); got.Move.State != before.Move.State || !got.Move.UpdatedAt.Equal(before.Move.UpdatedAt) {
		t.Fatalf("switch off: the record changed: %+v", got.Move)
	}
	if c := h.claim(t); c.Gen != claimBefore.Gen {
		t.Fatalf("switch off: the claim was written (gen %d → %d)", claimBefore.Gen, c.Gen)
	}
	if r, _ := readSourceReport(context.Background(), h.s, before); r != nil {
		t.Fatal("switch off: a source report was written")
	}
	if n := strings.Count(logs(), "is off on this box"); n != 1 {
		t.Fatalf("switch off: %d log lines, want 1:\n%s", n, logs())
	}

	// eb.Strict: nil with the switch off, set with it on.
	prev := armedMovesWorker.Load()
	t.Cleanup(func() { armedMovesWorker.Store(prev) })
	for _, on := range []bool{false, true} {
		v := ""
		if on {
			v = "on"
		}
		t.Setenv(moves.SwitchEnv, v)
		eb := &activator.ExecBackend{}
		armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{}, ClaimProviders{}, nil, nil)
		if (eb.Strict != nil) != on {
			t.Fatalf("switch %v: eb.Strict set = %v", on, eb.Strict != nil)
		}
		if eb.Phase1 != on {
			t.Fatalf("switch %v: eb.Phase1 = %v", on, eb.Phase1)
		}
	}
}

// Without a deployment bucket there is no shared place for the staging
// lease (round-5 finding 11): the round writes S1 and does nothing else.
func TestPhase1WithoutADeploymentBucketRefuses(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.cfg.StagingLease, h.cfg.DeployBucket = nil, ""
	gen := h.claim(t).Gen
	h.pass(t)
	if n := len(h.f.stagingCalls()); n != 0 {
		t.Fatalf("%d staging requests", n)
	}
	if h.claim(t).Gen != gen {
		t.Fatal("the claim was written")
	}
	if sr := h.sourceReport(t); sr == nil || sr.S1.OK || sr.S1.Reason != s1NoDeployBucket {
		t.Fatalf("source report: %+v", sr)
	}
}

// Review Focus 2: the source's own lost-response PUT of move.json and a
// foreign object look alike, and the source may not read the lane. With
// the ledger lost and move.json present, the put answers ErrExists, the
// round records "present" and goes on, and nothing reads the target lane.
func TestStagingMoveJSONReplayAfterLostLedger(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "x", Size: 2}, "{}"})
	h.vs.deny = func(op, bucket, key string) error {
		if bucket == "data-uk" && strings.HasPrefix(key, "moves/") {
			return errors.New("AccessDenied")
		}
		return nil
	}
	h.pass(t)
	if got := h.ledger(t).MoveJSONVersion; got != moveJSONPresent {
		t.Fatalf("ledger move_json_version = %q, want %q", got, moveJSONPresent)
	}
	if keys := stagedKeys(h, moves.ReplicaPrefix(p1Host, p1ID)); len(keys) != 2 {
		t.Fatalf("the round did not go on: %v", keys)
	}
	if calls := h.vs.callsOn(moves.StagingPrefix(p1Host, p1ID)); len(calls) != 0 {
		t.Fatalf("the source read the target lane: %v", calls)
	}
}

// rewriteClaimOp edits p1Host's claim op in place, as another writer would.
func rewriteClaimOp(t *testing.T, s release.Store, edit func(*hostclaim.Op)) {
	t.Helper()
	obs, err := hostclaim.Read(context.Background(), s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(context.Background(), s, obs, func(c *hostclaim.Claim) error { edit(&c.Op); return nil }); err != nil {
		t.Fatal(err)
	}
}

// Every transition is guarded by op identity: a replacement move written
// between the round's read and its CAS is ErrMoveConflict, and the round
// ends without advancing it.
func TestRoundNeverAdvancesAReplacementMove(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	h.start(t, "auto")
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	// Between the next round's read and its promotion, the record's move is
	// replaced by another in the same state.
	h.f.failAt = func(op, key string) error {
		h.f.failAt = nil
		_, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
			rec.Move.ID = p1ID2
			return nil
		})
		return err
	}
	h.f.putSized("data-ie", "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000003-0000000000000003.ltx", "cur-c", 10)
	h.pass(t)
	got := h.read(t, p1Host)
	if got.Move.ID != p1ID2 || got.Move.State != provision.MovePreparing || got.Move.Inputs != nil {
		t.Fatalf("the replacement move was advanced: %+v", got.Move)
	}
}

// The source copy is bound to the claim's location and generation (round-3
// finding 5): from a scoped location with two generations only g-cur's
// level files reach the lane; from a legacy root location with g-x/ beside
// its level directories, only the level directories do.
func TestReplicaSourceSelectsTheClaimsGeneration(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	c := h.claim(t)
	prefix, _ := moves.ReplicaSource(*c.Holder)
	if want := "instances/" + p1Host + "/" + h.inc + "/g-cur/"; prefix != want {
		t.Fatalf("prefix %q, want %q", prefix, want)
	}
	h.pass(t)
	got := stagedKeys(h, moves.ReplicaPrefix(p1Host, p1ID))
	if want := []string{"0000/0000000000000001-0000000000000002.ltx", "0001/0000000000000001-0000000000000002.ltx"}; strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("scoped location staged %v, want %v", got, want)
	}

	// The legacy root location.
	h2 := newPhase1Harness(t)
	h2.start(t, "manual")
	rewriteP1Claim(t, h2.s, func(c *hostclaim.Claim) {
		c.Holder.Location, c.Holder.Generation = "instances/"+p1Host, hostclaim.RootGeneration
	})
	root := "instances/" + p1Host + "/"
	h2.f.putSized("data-ie", root+"0000/0000000000000001-0000000000000005.ltx", "root-a", 10)
	h2.f.putSized("data-ie", root+"0009/0000000000000001-0000000000000005.ltx", "root-snap", 10)
	h2.f.putSized("data-ie", root+"g-x/0000/0000000000000001-0000000000000001.ltx", "gx", 10)
	h2.pass(t)
	got = stagedKeys(h2, moves.ReplicaPrefix(p1Host, p1ID))
	if want := []string{"0000/0000000000000001-0000000000000005.ltx", "0009/0000000000000001-0000000000000005.ltx"}; strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("root location staged %v, want %v", got, want)
	}
}

func rewriteP1Claim(t *testing.T, s release.Store, edit func(*hostclaim.Claim)) {
	t.Helper()
	obs, err := hostclaim.Read(context.Background(), s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(context.Background(), s, obs, func(c *hostclaim.Claim) error { edit(c); return nil }); err != nil {
		t.Fatal(err)
	}
}

// Every staging write goes through the guard: with the lease too close to
// its end, the round issues no staging request of any class (move.json,
// the replica copy, the archive) and nothing reaches the target bucket.
// And for every point at which the lease turns unsafe, the requests issued
// are exactly the checks that passed: no class signs without one.
func TestEveryStagingWriteIsGuarded(t *testing.T) {
	t.Parallel()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.lease.until = h.clock.now.Add(moves.SignatureLife) // inside SignatureLife + skew
	h.pass(t)
	if calls := h.f.stagingCalls(); len(calls) != 0 {
		t.Fatalf("guarded requests issued past the bound: %+v", calls)
	}
	if keys := stagedKeys(h, "moves/"); len(keys) != 0 {
		t.Fatalf("the target bucket was written: %v", keys)
	}

	// A full first round with a safe lease: every class, each signed at
	// the guard's check time, every write to the target bucket guarded.
	full := newPhase1Harness(t)
	full.start(t, "manual")
	full.pass(t)
	calls := full.f.stagingCalls()
	ops := map[string]bool{}
	for _, c := range calls {
		ops[c.Op] = true
		if c.SignAt.IsZero() || c.Bucket != "data-uk" {
			t.Fatalf("unguarded or misdirected request: %+v", c)
		}
	}
	if !ops["put-if-absent"] || !ops["copy"] || !ops["put"] {
		t.Fatalf("not every class went through the guard: %v", ops)
	}
	if full.lease.reads != len(calls) {
		t.Fatalf("%d guard checks for %d requests", full.lease.reads, len(calls))
	}
	for _, op := range full.f.ops {
		if strings.Contains(op, "data-uk/") && !strings.HasPrefix(op, "copy ") {
			t.Fatalf("an unguarded write reached the target bucket: %s", op)
		}
	}
	for k := 1; k < len(calls); k++ {
		hk := newPhase1Harness(t)
		hk.start(t, "manual")
		hk.lease.unsafeAfter = k
		hk.pass(t)
		if n := len(hk.f.stagingCalls()); n != k {
			t.Fatalf("lease unsafe after %d checks: %d requests issued", k, n)
		}
	}
}

// The box-step refusals, written into S1 with nothing staged: a legacy
// record with the switch on (logged, never touched), a nested row, a
// sidecar environment file, every refused registry column, a non-instance
// or non-hibernating row (F14), and a bound private store.
func TestPreparingRefusesLegacyNestedAndRefusedColumns(t *testing.T) {
	ctx := context.Background()
	t.Run("legacy record", func(t *testing.T) {
		h := newPhase1Harness(t)
		h.record(t, p1Host, provision.MoveRequest("uk", time.Now()))
		logs := captureLog(t)
		h.pass(t)
		if len(h.f.ops) != 0 || len(h.f.stagingCalls()) != 0 || h.read(t, p1Host).Move.State != provision.MovePreparing {
			t.Fatalf("a legacy record was driven: %v", h.f.ops)
		}
		if !strings.Contains(logs(), legacyMoveSentence) {
			t.Fatalf("no MV-N7 log line:\n%s", logs())
		}
	})
	for name, tc := range map[string]struct {
		edit func(*phase1Harness)
		want string
	}{
		"nested":       {func(h *phase1Harness) { h.route.DBPath = filepath.Join(h.cfg.DataDir, "instances", "x", p1Host+".db") }, s1Nested},
		"sidecar":      {func(h *phase1Harness) { writeFileForTest(filepath.Join(h.cfg.DataDir, p1Host+"-sidecar.env"), "X=1") }, s1Sidecar(p1Host)},
		"not instance": {func(h *phase1Harness) { h.route.Kind = "service" }, s1NotExec(p1Host)},
		"platform":     {func(h *phase1Harness) { h.route.Platform = true }, s1NotExec(p1Host)},
		"no hibernate": {func(h *phase1Harness) { h.route.Hibernate = false }, s1NoHibernate(p1Host)},
		"caps":         {func(h *phase1Harness) { h.route.Caps = "x" }, s1Column(p1Host, "caps")},
		"unit":         {func(h *phase1Harness) { h.route.Unit = "x" }, s1Column(p1Host, "unit")},
		"backing":      {func(h *phase1Harness) { h.route.Backing = "unit" }, s1Column(p1Host, "backing")},
		"tunnel":       {func(h *phase1Harness) { h.route.Tunnel = "x" }, s1Column(p1Host, "tunnel")},
		"fleet box":    {func(h *phase1Harness) { h.route.FleetBox = "x" }, s1Column(p1Host, "fleet_box")},
		"pool":         {func(h *phase1Harness) { h.route.Pool = "x" }, s1Column(p1Host, "pool")},
		"canary":       {func(h *phase1Harness) { h.route.Canary = true }, s1Column(p1Host, "canary")},
		"addr":         {func(h *phase1Harness) { h.route.Addr = "x:1" }, s1Column(p1Host, "addr")},
		"addrs":        {func(h *phase1Harness) { h.route.Addrs = "x:1" }, s1Column(p1Host, "addrs")},
		"health":       {func(h *phase1Harness) { h.route.Health = "/h" }, s1Column(p1Host, "health")},
		"replica":      {func(h *phase1Harness) { h.route.Replica = true }, s1Column(p1Host, "replica")},
		"replicas":     {func(h *phase1Harness) { h.route.Replicas = "uk" }, s1Column(p1Host, "replicas")},
		"private storage": {func(h *phase1Harness) {
			// The record has no Environment, so its store hangs off its
			// slug (provision.InstanceStoreEnv).
			if _, err := h.s.Put(ctx, objstore.KeyIn("bab", "app", "s"), []byte(`{"environment":"s"}`), ""); err != nil {
				panic(err)
			}
		}, s1PrivateStorage(p1Host)},
	} {
		t.Run(name, func(t *testing.T) {
			h := newPhase1Harness(t)
			h.start(t, "manual")
			tc.edit(h)
			h.pass(t)
			if calls := h.f.stagingCalls(); len(calls) != 0 {
				t.Fatalf("staged despite the refusal: %+v", calls)
			}
			sr := h.sourceReport(t)
			if sr == nil || sr.S1.OK || sr.S1.Reason != tc.want {
				t.Fatalf("S1 = %+v, want %q", sr, tc.want)
			}
		})
	}
}

// K12 and T9: the claim's touched gains the target region and box before
// the first staging write, from a claim that touched neither.
func TestPreparingTouchesTheTargetRegionAndBoxFirst(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	if c := h.claim(t); containsString(c.Touched.Regions, "uk") || containsString(c.Touched.Boxes, "uk-box") {
		t.Fatalf("the fixture already touched uk: %+v", c.Touched)
	}
	var touched *hostclaim.Touched
	h.f.failAt = func(op, key string) error {
		if touched == nil {
			c := h.claim(t)
			touched = &c.Touched
		}
		return nil
	}
	h.pass(t)
	if touched == nil {
		t.Fatal("no staging write happened")
	}
	if !containsString(touched.Regions, "uk") || !containsString(touched.Boxes, "uk-box") {
		t.Fatalf("at the first staging write touched = %+v", *touched)
	}
}

// A failed copy is retried only by calling again: the lane asks the guard
// again before the retry, and a lease past the bound by then ends the round
// with no second request.
func TestStagingWriterRetriesWithANewCheck(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	// The first copy fails, and by its retry the lease is too close to
	// its end.
	copies := 0
	h.f.failAt = func(op, key string) error {
		if op != "copy" {
			return nil
		}
		copies++
		if copies == 1 {
			h.lease.until = h.clock.now.Add(time.Minute)
			return errors.New("injected 503")
		}
		return nil
	}
	logs := captureLog(t)
	h.pass(t)
	if copies != 1 {
		t.Fatalf("%d copy requests reached the store, want 1", copies)
	}
	if !strings.Contains(logs(), "retrying once") || !strings.Contains(logs(), moves.ErrLeaseUnsafe.Error()) {
		t.Fatalf("the retry did not go back to the guard:\n%s", logs())
	}

	// With the lease safe, the retry's request is signed at a new check.
	h2 := newPhase1Harness(t)
	h2.start(t, "manual")
	var signs []time.Time
	failed := false
	h2.f.failAt = func(op, key string) error {
		if op != "copy" {
			return nil
		}
		calls := h2.f.at
		signs = append(signs, calls[len(calls)-1].SignAt)
		if !failed {
			failed = true
			return errors.New("injected 503")
		}
		return nil
	}
	h2.pass(t)
	if len(signs) < 2 || !signs[1].After(signs[0]) {
		t.Fatalf("the retry reused the first signature time: %v", signs)
	}
}

// The source report is replaced each round (PutReplace), never refused as
// an existing object.
func TestSourceReportIsWrittenEachRound(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	first := h.sourceReport(t)
	h.probe = moves.Check{Reason: s1NoAnswer(p1Host)}
	h.pass(t)
	second := h.sourceReport(t)
	if first == nil || second == nil || !second.At.After(first.At) || second.S1.OK || second.S1.Reason != s1NoAnswer(p1Host) {
		t.Fatalf("first %+v second %+v", first, second)
	}
}

// The archive is rebuilt only when the plaintext digest changes.
func TestArchiveIsRebuiltOnlyWhenPlainChanges(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	sealedPuts := func() int {
		n := 0
		for _, c := range h.f.stagingCalls() {
			if c.Op == "put" && strings.HasSuffix(c.Key, ".sealed") {
				n++
			}
		}
		return n
	}
	h.pass(t)
	h.pass(t)
	if n := sealedPuts(); n != 1 {
		t.Fatalf("%d archives staged over two unchanged rounds, want 1", n)
	}
	first := h.ledger(t).ArchivePlain
	if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed"), 0o644); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	if n := sealedPuts(); n != 2 {
		t.Fatalf("%d archives staged after a change, want 2", n)
	}
	if l := h.ledger(t); l.ArchivePlain == first || l.ArchiveVersions == nil || l.ArchiveVersions.SealedVersion == "" || l.ArchiveVersions.ManifestVersion == "" {
		t.Fatalf("ledger after the change: %+v", l)
	}
	// A file that changes under the walk (archive.ErrChanged), while the
	// digest is taken or while the archive is built, after an earlier
	// archive passed T5: the round ends before readiness, publishes no
	// source report, and never promotes on the archive the target checked
	// before (Task 14 review finding 2).
	for _, stage := range []string{"digest", "build"} {
		t.Run("changed during the "+stage, func(t *testing.T) {
			h2 := newPhase1Harness(t)
			h2.start(t, "manual")
			h2.pass(t)
			h2.targetPass(t, h2.targetCfg())
			if f := failing(h2.targetReport(t)); f[moves.CheckArchive] != "" {
				t.Fatalf("the first archive did not pass T5: %v", f)
			}
			before := h2.sourceReport(t)
			if err := os.WriteFile(filepath.Join(h2.cfg.DataDir, p1Host, "upload.txt"), []byte("changing"), 0o644); err != nil {
				t.Fatal(err)
			}
			changed := fmt.Errorf("%w: upload.txt", archive.ErrChanged)
			prevDigest, prevBuild := archivePlainDigest, archiveBuild
			t.Cleanup(func() { archivePlainDigest, archiveBuild = prevDigest, prevBuild })
			if stage == "digest" {
				archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, changed }
			} else {
				archiveBuild = func(string, string, io.Writer) (string, []archive.Entry, int64, error) { return "", nil, 0, changed }
			}
			h2.pass(t)
			if after := h2.sourceReport(t); !after.At.Equal(before.At) {
				t.Fatalf("a retried round published a source report (%v, was %v)", after.At, before.At)
			}
			if got := h2.read(t, p1Host).Move.State; got != provision.MovePreparing {
				t.Fatalf("a retried round promoted on the old archive: %s", got)
			}
			if l := h2.ledger(t); l.ArchivePlain != before.FilesPlain {
				t.Fatalf("the ledger moved to %s", l.ArchivePlain)
			}
			archivePlainDigest, archiveBuild = prevDigest, prevBuild
			h2.pass(t)
			if sr := h2.sourceReport(t); sr.FilesPlain == before.FilesPlain || h2.read(t, p1Host).Move.State != provision.MovePreparing {
				t.Fatalf("after the retry: files %s, state %s", sr.FilesPlain, h2.read(t, p1Host).Move.State)
			}
		})
	}

	// An archive this move can never carry is a standing S1 refusal.
	if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	if sr := h.sourceReport(t); sr.S1.OK || !strings.Contains(sr.S1.Reason, "keeps a file this move cannot copy") {
		t.Fatalf("S1 = %+v", sr.S1)
	}
}

// cut == "auto": from ready the source itself marks the cut requested, for
// a holder at the root location (the only one the cut accepts, ruling
// T5-lease; TestScopedAutomaticMoveDoesNotLoop covers the other).
func TestAutoCutMarksCuttingFromReady(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "auto")
	rootHolder(t, h)
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	h.pass(t)
	got := h.read(t, p1Host)
	if got.Move.State != provision.MoveCutting || got.Move.CutPhase != provision.CutRequested || got.Move.Inputs == nil {
		t.Fatalf("auto cut: %+v", got.Move)
	}
	// A manual move stays ready.
	h2 := newPhase1Harness(t)
	h2.start(t, "manual")
	h2.pass(t)
	h2.targetPass(t, h2.targetCfg())
	h2.pass(t)
	h2.pass(t)
	if got := h2.read(t, p1Host).Move; got.State != provision.MoveReady {
		t.Fatalf("manual cut: %+v", got)
	}
}

// Promotion pins only the release the target checked (Task 14 review
// finding 3): after the target checks v7, a source that moves to v8 does
// not promote until the target has checked v8, and then pins v8.
func TestPromotionNeedsTheReleaseTheTargetChecked(t *testing.T) {
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	if r := h.targetReport(t); r.Inputs.Release != "v7" {
		t.Fatalf("the target checked %q", r.Inputs.Release)
	}
	pointLiveForTest(t, h.optDir, p1Host, "v8")
	h.pass(t)
	if got := h.read(t, p1Host).Move; got.State != provision.MovePreparing || got.Inputs != nil {
		t.Fatalf("promoted with v8 pinned on a v7 check: %+v", got)
	}
	h.targetPass(t, h.targetCfg())
	h.pass(t)
	if got := h.read(t, p1Host).Move; got.State != provision.MoveReady || got.Inputs == nil || got.Inputs.Release != "v8" {
		t.Fatalf("after the target checked v8: %+v", got)
	}
}

// S1 asks the path the instances manifest declares. A manifest that cannot
// be read fails S1, even though "/" would answer (Task 14 review finding
// 5); "/" is used only when the manifest was read and declares no path.
func TestS1ReadsTheDeclaredReadinessPath(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	if sr := h.sourceReport(t); !sr.S1.OK || len(h.probePaths) != 1 || h.probePaths[0] != "/" {
		t.Fatalf("no declared path: S1 %+v, probed %v", sr.S1, h.probePaths)
	}
	if err := provision.PutManifest(ctx, h.s, "bab", "app", provision.Manifest{Schema: 1, Moves: &provision.MovesManifest{ReadyPath: "/health"}}); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	if got := h.probePaths[len(h.probePaths)-1]; got != "/health" {
		t.Fatalf("probed %q, want the declared /health", got)
	}
	// The manifest no longer parses: S1 fails, and "/" is never asked.
	_, etag, err := h.s.Get(ctx, provision.ManifestKey("bab", "app"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := h.s.Put(ctx, provision.ManifestKey("bab", "app"), []byte("{not json"), etag); err != nil {
		t.Fatal(err)
	}
	asked := len(h.probePaths)
	h.pass(t)
	if sr := h.sourceReport(t); sr.S1.OK || sr.S1.Reason != s1NoReadyPath {
		t.Fatalf("S1 with an unreadable manifest: %+v", sr.S1)
	}
	if len(h.probePaths) != asked {
		t.Fatalf("the instance was probed at %v with no readable manifest", h.probePaths[asked:])
	}
}

// Archive errors reach S1 only as recorded sentences, chosen by errors.Is
// and the path the error carries as data; the raw error text, which may
// name anything, goes to the log alone (ruling T15-sinks).
func TestArchiveErrorsReachS1AsRecordedSentences(t *testing.T) {
	secret := "EIO reading /data/other-tenant/secret"
	for name, tc := range map[string]struct {
		err  error
		want string
	}{
		"special file":      {&archive.PathError{Err: archive.ErrSpecialFile, Path: "link"}, s1SpecialFile(p1Host, "link")},
		"over cap":          {fmt.Errorf("walk: %w", archive.ErrOverCap), s1OverCap(p1Host)},
		"too many":          {archive.ErrTooManyEntries, s1TooManyEntries},
		"long path":         {&archive.PathError{Err: archive.ErrBadPath, Path: "deep"}, s1LongPath("deep")},
		"special, no path":  {fmt.Errorf("%w: %s", archive.ErrSpecialFile, secret), s1ArchiveFailed},
		"bad path, no path": {fmt.Errorf("%w: %s", archive.ErrBadPath, secret), s1ArchiveFailed},
		"unrecognised":      {errors.New(secret), s1ArchiveFailed},
	} {
		t.Run(name, func(t *testing.T) {
			h := newPhase1Harness(t)
			h.start(t, "manual")
			prev := archivePlainDigest
			t.Cleanup(func() { archivePlainDigest = prev })
			archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, tc.err }
			logs := captureLog(t)
			h.pass(t)
			sr := h.sourceReport(t)
			if sr == nil || sr.S1.OK || sr.S1.Reason != tc.want {
				t.Fatalf("S1 = %+v, want %q", sr, tc.want)
			}
			if strings.Contains(sr.S1.Reason, "EIO") || strings.Contains(sr.S1.Reason, "other-tenant") {
				t.Fatalf("the raw error reached S1: %q", sr.S1.Reason)
			}
			if !strings.Contains(logs(), tc.err.Error()) {
				t.Fatalf("the raw error was not logged:\n%s", logs())
			}
		})
	}
}

// MV-N79 and MV-N81 are the archive's own texts, written from data.
func TestArchiveSentencesMatchTheArchiveRows(t *testing.T) {
	if s1TooManyEntries != archive.ErrTooManyEntries.Error() {
		t.Fatalf("%q vs %q", s1TooManyEntries, archive.ErrTooManyEntries.Error())
	}
	// A relative path over MaxPathLen, made one directory at a time
	// through os.Root so no single call meets the OS path limit.
	files := t.TempDir()
	long := strings.Repeat("d", 200)
	cur, err := os.OpenRoot(files)
	if err != nil {
		t.Fatal(err)
	}
	for depth := 0; depth*(len(long)+1) <= archive.MaxPathLen; depth++ {
		if err := cur.Mkdir(long, 0o755); err != nil {
			t.Fatal(err)
		}
		next, err := cur.OpenRoot(long)
		cur.Close()
		if err != nil {
			t.Fatal(err)
		}
		cur = next
	}
	cur.Close()
	_, _, _, err = archive.PlainDigest(files, "")
	var pe *archive.PathError
	if !errors.As(err, &pe) || !errors.Is(err, archive.ErrBadPath) || err.Error() != s1LongPath(pe.Path) {
		t.Fatalf("err %v; s1LongPath %q", err, s1LongPath(pe.Path))
	}
}

// readyAuto drives a move to a persisted ready (cut manual, so the
// promotion does not cut in the same pass), then makes it automatic: the
// state a crash between the promotion and the cut leaves behind.
func readyAuto(t *testing.T) *phase1Harness {
	t.Helper()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	rootHolder(t, h) // an automatic cut is requested only for a root holder (ruling T5-lease)
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MoveReady {
		t.Fatalf("the fixture did not reach ready: %s", got)
	}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.Cut = "auto"
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	return h
}

// The re-review's regression: from a persisted ready, automatic move, a
// round whose archive fails with an unrecognised error publishes MV-N151
// and requests no cut. Without the failure, the same round does.
func TestAnUnknownArchiveErrorRequestsNoCut(t *testing.T) {
	h := readyAuto(t)
	prev := archivePlainDigest
	t.Cleanup(func() { archivePlainDigest = prev })
	archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, errors.New("EIO") }
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MoveReady {
		t.Fatalf("an unknown archive error requested the cut: %s", got)
	}
	if sr := h.sourceReport(t); sr.S1.OK || sr.S1.Reason != s1ArchiveFailed {
		t.Fatalf("S1 = %+v", sr.S1)
	}
	archivePlainDigest = prev
	h.pass(t)
	if got := h.read(t, p1Host).Move; got.State != provision.MoveCutting {
		t.Fatalf("a clean round did not request the cut: %+v", got)
	}
}

// The invariant: a round requests a cut only when this round's own source
// report passed every check and was published. A failure injected into
// each source check, retry or error of the round leaves a ready,
// automatic move ready.
func TestOnlyAPassingRoundRequestsTheCut(t *testing.T) {
	ctx := context.Background()
	for name, inject := range map[string]func(t *testing.T, h *phase1Harness){
		"none (the control)": nil,
		"S1 the instance does not answer": func(t *testing.T, h *phase1Harness) {
			h.probe = moves.Check{Reason: s1NoAnswer(p1Host)}
		},
		"S1 the manifest cannot be read": func(t *testing.T, h *phase1Harness) {
			_, etag, _ := h.s.Get(ctx, provision.ManifestKey("bab", "app"))
			if _, err := h.s.Put(ctx, provision.ManifestKey("bab", "app"), []byte("{"), etag); err != nil {
				t.Fatal(err)
			}
		},
		"S1 no probe on this box": func(t *testing.T, h *phase1Harness) { h.cfg.ReadyProbe = nil },
		"S1 no secrets key":       func(t *testing.T, h *phase1Harness) { h.cfg.SecretsKey = "" },
		"S1 a refused column":     func(t *testing.T, h *phase1Harness) { h.route.Canary = true },
		"S1 a special file": func(t *testing.T, h *phase1Harness) {
			if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
				t.Fatal(err)
			}
		},
		"S2 the next spawn is not strict": func(t *testing.T, h *phase1Harness) {
			h.strict = moves.Check{Reason: "S2: the next spawn of this host is not strict"}
		},
		"S2 no strict hook": func(t *testing.T, h *phase1Harness) { h.cfg.StrictNext = nil },
		"the archive changed (retry)": func(t *testing.T, h *phase1Harness) {
			prev := archivePlainDigest
			t.Cleanup(func() { archivePlainDigest = prev })
			archivePlainDigest = func(string, string) (string, int64, int, error) { return "", 0, 0, archive.ErrChanged }
		},
		"the launch environment cannot be composed": func(t *testing.T, h *phase1Harness) {
			h.cfg.LaunchEnv = func(string) (map[string]string, map[string]bool, error) { return nil, nil, errors.New("no row") }
		},
		"a staging write fails": func(t *testing.T, h *phase1Harness) {
			h.f.putSized("data-ie", "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000003-0000000000000003.ltx", "new", 10)
			h.f.failAt = func(string, string) error { return errors.New("injected 503") }
		},
		"the staging lease is too close to its end": func(t *testing.T, h *phase1Harness) {
			h.f.putSized("data-ie", "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000003-0000000000000003.ltx", "new", 10)
			h.lease.until = h.clock.now.Add(time.Minute)
		},
		"no deployment bucket": func(t *testing.T, h *phase1Harness) { h.cfg.StagingLease = nil },
		"the claim is not the move's": func(t *testing.T, h *phase1Harness) {
			rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = hostclaim.PhaseAborting })
		},
	} {
		t.Run(name, func(t *testing.T) {
			h := readyAuto(t)
			if inject != nil {
				inject(t, h)
			}
			h.pass(t)
			got := h.read(t, p1Host).Move.State
			if inject == nil && got != provision.MoveCutting {
				t.Fatalf("the control round did not request the cut: %s", got)
			}
			if inject != nil && got != provision.MoveReady {
				t.Fatalf("a failing round requested the cut: %s", got)
			}
		})
	}
}

// One strict setter (ruling F34/G9): a host is strict on every box that is a
// participant of its Phase 1 move. The source stays strict after the flip
// (spec §5.3): its record's region is the target by then, but Move.From
// still names it until the close CAS clears the move.
func TestSetStrictCoversEveryParticipant(t *testing.T) {
	const host = "a.go.example.com"
	move := func(id, from, to string) *provision.MoveState {
		return &provision.MoveState{ID: id, From: from, To: to, State: provision.MoveCutting}
	}
	cases := []struct {
		name   string
		box    string
		region string // the record's region
		move   *provision.MoveState
		want   bool
	}{
		{"source before the flip", "ie", "ie", move(p1ID, "ie", "ca"), true},
		{"source after the flip", "ie", "ca", move(p1ID, "ie", "ca"), true},
		{"target before the flip", "ca", "ie", move(p1ID, "ie", "ca"), true},
		{"target after the flip", "ca", "ca", move(p1ID, "ie", "ca"), true},
		{"a box that is neither", "us", "ca", move(p1ID, "ie", "ca"), false},
		{"the source once the move closed", "ie", "ca", nil, false},
		{"a legacy move names no one", "ie", "ca", move("", "ie", "ca"), false},
		{"a legacy move on the target", "ca", "ie", move("", "ie", "ca"), false},
		{"a record with no move", "ie", "ie", nil, false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			w := &movesWorker{cfg: moveCfg{Region: tc.box}}
			w.setStrict([]provision.Record{{Host: host, Region: tc.region, Move: tc.move}})
			if got := w.strictHost(host); got != tc.want {
				t.Fatalf("strictHost = %v, want %v", got, tc.want)
			}
			if w.strictHost("other.example.com") {
				t.Fatal("an unrelated host is strict")
			}
		})
	}
}