package moves

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"os"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// passingTarget is a passing report made at `at` with real evidence: a
// 100 MiB database priced from rates measured a minute earlier, in a 120 s
// window. goodSource is the matching source report.
func passingTarget(at time.Time) *TargetReport {
	r := &TargetReport{V: 1, At: at, Checks: map[string]Check{}, CutBudgetS: 120, Inputs: provision.MoveInputs{Release: "v7"}}
	for _, id := range AllChecks {
		r.Checks[id] = Check{OK: true}
	}
	m := Measurements{
		Cross: MeasuredRate{MiBPerS: 10, At: at.Add(-time.Minute)},
		S3:    MeasuredRate{MiBPerS: 250, At: at.Add(-time.Minute)},
		Disk:  MeasuredRate{MiBPerS: 500, At: at.Add(-time.Minute)},
		Fsync: FilesRate{PerS: 1600, At: at.Add(-time.Minute)},
	}
	res := BudgetCheck(100, 0, m, 16<<20, 120, at)
	if !res.Check.OK {
		panic("the test fixture does not fit its own window: " + res.Check.Reason)
	}
	r.RecordBudget(100, 0, m, res)
	return r
}

func sourceWith(now time.Time, s1, s2 Check) *SourceReport {
	sr := goodSource(now)
	sr.S1, sr.S2 = s1, s2
	return sr
}

func goodSource(now time.Time) *SourceReport {
	return &SourceReport{V: 1, At: now, S1: Check{OK: true}, S2: Check{OK: true}, CrossRate: 10, CrossRateAt: now.Add(-time.Minute), Release: "v7"}
}

func TestReadyNeedsFreshReports(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	pass := passingTarget
	src := goodSource(now)
	if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-time.Minute)), src, now); !ok {
		t.Fatal(why)
	}
	// Review Focus 5: a passing report older than 10 minutes does not promote.
	if ok, why := Ready(1<<20, 16<<20, pass(now.Add(-11*time.Minute)), src, now); ok || !strings.Contains(why, "stale") {
		t.Fatalf("stale report promoted: %v %s", ok, why)
	}
	// The same bound on the source's report.
	oldSrc := goodSource(now)
	oldSrc.At = now.Add(-11 * time.Minute)
	if ok, why := Ready(0, 16<<20, pass(now), oldSrc, now); ok || !strings.Contains(why, "stale") {
		t.Fatalf("stale source report promoted: %v %s", ok, why)
	}
	// Over the bar, a failed check, a failed S1, a missing report, zero attempts.
	if ok, _ := Ready(17<<20, 16<<20, pass(now), src, now); ok {
		t.Fatal("over the bar")
	}
	bad := pass(now)
	bad.Checks[CheckBudget] = Check{Reason: "needs 131 s, window 120 s"}
	if ok, why := Ready(0, 16<<20, bad, src, now); ok || !strings.Contains(why, "T11") {
		t.Fatalf("failed check: %v %s", ok, why)
	}
	if ok, _ := Ready(0, 16<<20, pass(now), sourceWith(now, Check{Reason: "503"}, Check{OK: true}), now); ok {
		t.Fatal("failed S1")
	}
	if ok, why := Ready(0, 16<<20, pass(now), sourceWith(now, Check{OK: true}, Check{Reason: "strict spawn refused"}), now); ok || !strings.Contains(why, "S2") {
		t.Fatalf("failed S2: %v %s", ok, why)
	}
	if ok, _ := Ready(0, 16<<20, nil, src, now); ok {
		t.Fatal("no report")
	}
	if ok, _ := Ready(0, 16<<20, pass(now), nil, now); ok {
		t.Fatal("no source report")
	}
	none := pass(now)
	none.AttemptsAdmitted = 0
	if ok, _ := Ready(0, 16<<20, none, src, now); ok {
		t.Fatal("zero attempts admitted")
	}
}

// Review Focus 5, second half: a stale report after ready demotes nothing.
// Ready only ever gates promotion. The one transition it feeds is
// MovePromoteReady, which is legal from preparing alone, so a move that is
// already ready stays ready whatever the next reading of Ready says, and
// reading Ready changes neither report.
func TestStaleReportAfterReadyDemotesNothing(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	target := passingTarget(now)
	src := goodSource(now)
	rec := &provision.Record{Move: &provision.MoveState{State: provision.MovePreparing}}

	if ok, why := Ready(0, 16<<20, target, src, now); !ok {
		t.Fatal(why)
	}
	if err := provision.MovePromoteReady(now)(rec); err != nil {
		t.Fatal(err)
	}
	if rec.Move.State != provision.MoveReady {
		t.Fatalf("not promoted: %s", rec.Move.State)
	}

	// Both reports go stale. The reading is false, and nothing moves.
	targetBefore, srcBefore := *target, *src
	later := now.Add(ReportFreshness + time.Minute)
	if ok, why := Ready(0, 16<<20, target, src, later); ok || !strings.Contains(why, "stale") {
		t.Fatalf("stale reading: %v %s", ok, why)
	}
	if !reflect.DeepEqual(*target, targetBefore) || !reflect.DeepEqual(*src, srcBefore) {
		t.Fatal("Ready changed a report")
	}
	// The host acts on a true reading only, so the stale reading does not
	// reach the journal; and if it did, there is no way back to preparing.
	if err := provision.MovePromoteReady(later)(rec); err == nil {
		t.Fatal("a second promotion was accepted")
	}
	if rec.Move.State != provision.MoveReady {
		t.Fatalf("a stale report demoted the move to %s", rec.Move.State)
	}
}

func TestReadyRejectsMismatchedInputs(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	mk := func(targetFiles, sourceFiles, targetEnv, sourceEnv string) (*TargetReport, *SourceReport) {
		tr := passingTarget(now)
		tr.Inputs = provision.MoveInputs{FilesPlain: targetFiles, EnvReport: targetEnv, Release: "v7"}
		sr := goodSource(now)
		sr.FilesPlain, sr.Env = sourceFiles, EnvReport{V: 1, Digest: sourceEnv}
		return tr, sr
	}
	tr, sr := mk("sha256:aaa", "sha256:aaa", "env-1", "env-1")
	if ok, why := Ready(0, 16<<20, tr, sr, now); !ok {
		t.Fatalf("equal inputs: %s", why)
	}
	tr, sr = mk("sha256:aaa", "sha256:bbb", "env-1", "env-1")
	ok, why := Ready(0, 16<<20, tr, sr, now)
	if ok || !strings.Contains(why, "sha256:aaa") || !strings.Contains(why, "sha256:bbb") {
		t.Fatalf("files mismatch: %v %q", ok, why)
	}
	tr, sr = mk("sha256:aaa", "sha256:aaa", "env-1", "env-2")
	ok, why = Ready(0, 16<<20, tr, sr, now)
	if ok || !strings.Contains(why, "env-1") || !strings.Contains(why, "env-2") {
		t.Fatalf("env mismatch: %v %q", ok, why)
	}
}

func TestReadyRefusesAnEstimateMadeUnderASmallerMirrorBar(t *testing.T) {
	now := t11Now
	src := goodSource(now)
	// made returns a report whose budget was priced for the given bar.
	made := func(barBytes int64) *TargetReport {
		tr := passingTarget(now)
		m := Measurements{
			Cross: MeasuredRate{MiBPerS: 10, At: now.Add(-time.Minute)},
			S3:    MeasuredRate{MiBPerS: 250, At: tr.S3At},
			Disk:  MeasuredRate{MiBPerS: 500, At: tr.DiskAt},
			Fsync: FilesRate{PerS: 1600, At: tr.FsyncAt},
		}
		res := BudgetCheck(100, 0, m, barBytes, 120, now)
		if !res.Check.OK {
			t.Fatalf("%+v", res)
		}
		tr.RecordBudget(100, 0, m, res)
		return tr
	}
	// Priced at the 16 MiB default, the operator's bar raised to 32 MiB:
	// the estimate is too small for it.
	if ok, why := Ready(0, 32<<20, made(16<<20), src, now); ok || !strings.Contains(why, "T11") || !strings.Contains(why, "priced") {
		t.Fatalf("under-priced estimate promoted: %v %q", ok, why)
	}
	// Priced at 32 MiB: fine at 32, and fine at a lower bar.
	if ok, why := Ready(0, 32<<20, made(32<<20), src, now); !ok {
		t.Fatal(why)
	}
	if ok, why := Ready(0, 8<<20, made(32<<20), src, now); !ok {
		t.Fatal(why)
	}
	// Priced at 8, bar 16: too small.
	if ok, _ := Ready(0, 16<<20, made(8<<20), src, now); ok {
		t.Fatal("a smaller priced bar promoted")
	}
	// A mirror bar that is not finite never promotes, nor a negative one.
	for _, bar := range []float64{math.Inf(1), math.NaN(), -16} {
		tr := made(16 << 20)
		tr.Rates.MirrorBar = bar
		if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
			t.Fatalf("mirror bar %v promoted", bar)
		}
	}
}

// The controller's fail-closed ruling: Ready re-checks the evidence behind
// the budget at its own `now`, whatever the report's checks say.
func TestReadyValidatesTheEvidenceItself(t *testing.T) {
	now := t11Now
	if ok, why := Ready(0, 16<<20, passingTarget(now), goodSource(now), now); !ok {
		t.Fatal(why)
	}
	bad := []float64{0, -1, math.NaN(), math.Inf(1), math.Inf(-1)}
	type c struct {
		name   string
		mutate func(*TargetReport, *SourceReport)
	}
	var cases []c
	for _, v := range bad {
		v := v
		cases = append(cases,
			c{fmt.Sprintf("s3 rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.S3 = v }},
			c{fmt.Sprintf("disk rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Disk = v }},
			c{fmt.Sprintf("cross rate in the source %v", v), func(_ *TargetReport, sr *SourceReport) { sr.CrossRate = v }},
			c{fmt.Sprintf("cross rate in the target %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.Cross = v }},
			c{fmt.Sprintf("fsync rate %v", v), func(tr *TargetReport, _ *SourceReport) { tr.Rates.FsyncRate = v }},
		)
	}
	expired := now.Add(-MeasurementFreshness - time.Second)
	cases = append(cases,
		c{"s3 time missing", func(tr *TargetReport, _ *SourceReport) { tr.S3At = time.Time{} }},
		c{"s3 expired", func(tr *TargetReport, _ *SourceReport) { tr.S3At = expired }},
		c{"s3 from the future", func(tr *TargetReport, _ *SourceReport) { tr.S3At = now.Add(time.Hour) }},
		c{"disk time missing", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = time.Time{} }},
		c{"disk expired", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }},
		c{"disk from the future", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = now.Add(time.Hour) }},
		c{"fsync time missing", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = time.Time{} }},
		c{"fsync expired", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }},
		c{"fsync from the future", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = now.Add(time.Hour) }},
		c{"entries negative", func(tr *TargetReport, _ *SourceReport) { tr.Entries = -1 }},
		c{"too many entries for the window", func(tr *TargetReport, _ *SourceReport) { tr.Entries = 1 << 30 }},
		c{"cross time missing", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = time.Time{} }},
		c{"cross expired", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }},
		c{"cross from the future", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = now.Add(time.Hour) }},
		c{"source has measured no cross rate", func(_ *TargetReport, sr *SourceReport) { sr.CrossRate, sr.CrossRateAt = 0, time.Time{} }},
		c{"database size NaN", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = math.NaN() }},
		c{"database size negative", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = -1 }},
		c{"no window", func(tr *TargetReport, _ *SourceReport) { tr.CutBudgetS = 0 }},
		c{"database too big for the window", func(tr *TargetReport, _ *SourceReport) { tr.DBMiB = 1e6 }},
		c{"more attempts claimed than admitted", func(tr *TargetReport, _ *SourceReport) { tr.AttemptsAdmitted = ProbeAttemptsMax + 1 }},
		c{"no attempts claimed", func(tr *TargetReport, _ *SourceReport) { tr.AttemptsAdmitted = 0 }},
	)
	for _, cs := range cases {
		tr, sr := passingTarget(now), goodSource(now)
		cs.mutate(tr, sr)
		ok, why := Ready(0, 16<<20, tr, sr, now)
		if ok || !strings.Contains(why, "T11") {
			t.Errorf("%s: ready=%v %q", cs.name, ok, why)
		}
	}

	// Which rate is named when a measurement is missing or expired.
	for _, w := range []struct {
		name   string
		mutate func(*TargetReport, *SourceReport)
		which  string
	}{
		{"s3", func(tr *TargetReport, _ *SourceReport) { tr.S3At = expired }, "regional S3"},
		{"disk", func(tr *TargetReport, _ *SourceReport) { tr.DiskAt = expired }, "the disk"},
		{"fsync", func(tr *TargetReport, _ *SourceReport) { tr.FsyncAt = expired }, "the disk's per-file sync"},
		{"cross", func(_ *TargetReport, sr *SourceReport) { sr.CrossRateAt = expired }, "the cross-region copy path"},
	} {
		tr, sr := passingTarget(now), goodSource(now)
		w.mutate(tr, sr)
		if _, why := Ready(0, 16<<20, tr, sr, now); !strings.Contains(why, "no fresh rate measurement for "+w.which) {
			t.Errorf("%s: %q", w.name, why)
		}
	}

	// A measurement that expires while its report is still fresh is
	// refused: the target's report is 5 minutes old, its rates 58 minutes
	// old when written, so an hour later than their time they are stale.
	tr := passingTarget(now.Add(-5 * time.Minute))
	tr.S3At, tr.DiskAt = now.Add(-58*time.Minute), now.Add(-58*time.Minute)
	if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
		t.Fatal(why)
	}
	if ok, why := Ready(0, 16<<20, tr, goodSource(now.Add(5*time.Minute)), now.Add(5*time.Minute)); ok || !strings.Contains(why, "T11") {
		t.Fatalf("a measurement that expired under a fresh report: %v %q", ok, why)
	}

	// A stored count at or below the recomputed one is accepted.
	tr = passingTarget(now)
	tr.AttemptsAdmitted = 1
	if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
		t.Fatal(why)
	}
}

func TestPassingNamesTheFirstFailureInIDOrder(t *testing.T) {
	r := passingTarget(time.Now())
	if ok, why := r.Passing(); !ok || why != "" {
		t.Fatalf("%v %q", ok, why)
	}
	r.Checks[CheckClaim] = Check{Reason: "touched"}
	r.Checks[CheckArchive] = Check{Reason: "no archive"}
	if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5: ") || !strings.Contains(why, "no archive") {
		t.Fatalf("%v %q", ok, why)
	}
	delete(r.Checks, CheckArchive)
	if ok, why := r.Passing(); ok || !strings.HasPrefix(why, "T5") || strings.Contains(why, "T9") {
		t.Fatalf("missing check: %v %q", ok, why)
	}
	// A failed check with no reason still says something.
	r.Checks[CheckArchive] = Check{}
	if ok, why := r.Passing(); ok || strings.TrimSpace(strings.TrimPrefix(why, "T5:")) == "" {
		t.Fatalf("empty reason: %v %q", ok, why)
	}
	// A report with no checks at all is not passing.
	if ok, _ := (TargetReport{}).Passing(); ok {
		t.Fatal("an empty report passed")
	}
}

func TestAllChecksListsT1ToT13(t *testing.T) {
	want := []string{"T1", "T2", "T3", "T4", "T5", "T6", "T7", "T8", "T9", "T10", "T11", "T12", "T13"}
	if !reflect.DeepEqual(AllChecks, want) {
		t.Fatalf("%v", AllChecks)
	}
	got := []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet, CheckArmed}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("%v", got)
	}
}

func TestReportsRoundTripWithFixedWireNames(t *testing.T) {
	at := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	tr := TargetReport{V: 1, Box: "uk-box", Region: "uk", MoveID: "mv_x", At: at,
		Checks:           map[string]Check{CheckProtocol: {OK: true}},
		Inputs:           provision.MoveInputs{FilesPlain: "p", EnvReport: "e"},
		AttemptsAdmitted: 2, CutBudgetS: 120, Rates: Rates{Cross: 10, S3: 250, Disk: 500, MirrorBar: 16},
		Notes: []string{"the carried .env overrides X"}}
	b, err := json.Marshal(tr)
	if err != nil {
		t.Fatal(err)
	}
	var m map[string]json.RawMessage
	if err := json.Unmarshal(b, &m); err != nil {
		t.Fatal(err)
	}
	for _, k := range []string{"v", "box", "region", "move_id", "at", "checks", "inputs", "attempts_admitted", "cut_budget_s", "rates", "notes"} {
		if _, ok := m[k]; !ok {
			t.Errorf("target report has no %q: %s", k, b)
		}
	}
	var back TargetReport
	if err := json.Unmarshal(b, &back); err != nil || !reflect.DeepEqual(back, tr) {
		t.Fatalf("%v %+v", err, back)
	}

	sr := SourceReport{V: 1, Box: "src", MoveID: "mv_x", At: at, S1: Check{OK: true}, S2: Check{OK: true},
		FilesPlain: "p", Release: "v1.2.3", CrossRate: 9.5, CrossRateAt: at,
		Archive: &ArchiveRef{ManifestKey: "a/m", ManifestVersion: "v1", SealedObjectKey: "a/s", SealedVersion: "v2", Plain: "p"}}
	b, err = json.Marshal(sr)
	if err != nil {
		t.Fatal(err)
	}
	m = nil
	if err := json.Unmarshal(b, &m); err != nil {
		t.Fatal(err)
	}
	for _, k := range []string{"v", "box", "move_id", "at", "env", "s1", "s2", "files_plain", "release", "archive", "cross_rate", "cross_rate_at"} {
		if _, ok := m[k]; !ok {
			t.Errorf("source report has no %q: %s", k, b)
		}
	}
	var ref map[string]string
	if err := json.Unmarshal(m["archive"], &ref); err != nil {
		t.Fatal(err)
	}
	for _, k := range []string{"manifest_key", "manifest_version", "sealed_object_key", "sealed_version", "plain"} {
		if _, ok := ref[k]; !ok {
			t.Errorf("archive ref has no %q", k)
		}
	}
	var sback SourceReport
	if err := json.Unmarshal(b, &sback); err != nil || !reflect.DeepEqual(sback, sr) {
		t.Fatalf("%v %+v", err, sback)
	}
	// No archive yet: the field is absent, not null.
	b, _ = json.Marshal(SourceReport{V: 1})
	if strings.Contains(string(b), `"archive"`) {
		t.Fatalf("an unpublished archive is written: %s", b)
	}
}

// ---- measured rates and T11 ----

var t11Now = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)

func goodMeasurements() Measurements {
	at := t11Now.Add(-time.Minute)
	return Measurements{
		Cross: MeasuredRate{MiBPerS: 10, At: at, Bytes: 100 << 20},
		S3:    MeasuredRate{MiBPerS: 250, At: at, Bytes: 16 << 20},
		Disk:  MeasuredRate{MiBPerS: 500, At: at, Bytes: 64 << 20},
		Fsync: FilesRate{PerS: 1600, At: at, Files: 512},
	}
}

func TestMeasurementsNeedFreshUsableRates(t *testing.T) {
	r, err := goodMeasurements().Rates(t11Now, 16<<20)
	if err != nil {
		t.Fatal(err)
	}
	if r.Cross != 10 || r.S3 != 250 || r.Disk != 500 || r.MirrorBar != 16 || r.FsyncRate != 1600 {
		t.Fatalf("%+v", r)
	}

	cases := []struct {
		name   string
		mutate func(m *Measurements)
		which  string
	}{
		{"missing cross", func(m *Measurements) { m.Cross = MeasuredRate{} }, "the cross-region copy path"},
		{"old s3", func(m *Measurements) { m.S3.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "regional S3"},
		{"future disk", func(m *Measurements) { m.Disk.At = t11Now.Add(time.Hour) }, "the disk"},
		{"zero rate", func(m *Measurements) { m.Disk.MiBPerS = 0 }, "the disk"},
		{"negative rate", func(m *Measurements) { m.S3.MiBPerS = -5 }, "regional S3"},
		{"NaN", func(m *Measurements) { m.Cross.MiBPerS = math.NaN() }, "the cross-region copy path"},
		{"+Inf", func(m *Measurements) { m.S3.MiBPerS = math.Inf(1) }, "regional S3"},
		{"no time", func(m *Measurements) { m.Disk.At = time.Time{} }, "the disk"},
		{"no fsync", func(m *Measurements) { m.Fsync = FilesRate{} }, "the disk's per-file sync"},
		{"old fsync", func(m *Measurements) { m.Fsync.At = t11Now.Add(-MeasurementFreshness - time.Second) }, "the disk's per-file sync"},
		{"NaN fsync", func(m *Measurements) { m.Fsync.PerS = math.NaN() }, "the disk's per-file sync"},
	}
	for _, c := range cases {
		m := goodMeasurements()
		c.mutate(&m)
		_, err := m.Rates(t11Now, 16<<20)
		if err == nil || !strings.Contains(err.Error(), "no fresh rate measurement for "+c.which) {
			t.Errorf("%s: %v", c.name, err)
		}
	}
	// Exactly an hour old is still fresh.
	m := goodMeasurements()
	m.S3.At = t11Now.Add(-MeasurementFreshness)
	if _, err := m.Rates(t11Now, 16<<20); err != nil {
		t.Fatal(err)
	}
}

func TestMeasuredRateFromBytesAndTime(t *testing.T) {
	m := RateOf(100<<20, 10*time.Second, t11Now)
	if m.MiBPerS != 10 || m.Bytes != 100<<20 || !m.At.Equal(t11Now) {
		t.Fatalf("%+v", m)
	}
	// No elapsed time or no bytes is no measurement, not an infinite rate.
	for _, m := range []MeasuredRate{RateOf(1<<20, 0, t11Now), RateOf(0, time.Second, t11Now), RateOf(1<<20, -time.Second, t11Now)} {
		if m.MiBPerS != 0 {
			t.Fatalf("%+v", m)
		}
	}
}

func TestSourceReportCarriesTheCrossRate(t *testing.T) {
	at := t11Now.Add(-time.Minute)
	if got := (&SourceReport{CrossRate: 9.5, CrossRateAt: at}).CrossMeasurement(); got.MiBPerS != 9.5 || !got.At.Equal(at) {
		t.Fatalf("%+v", got)
	}
	// A source that has measured nothing yields the zero measurement, which
	// Measurements.Rates refuses.
	m := goodMeasurements()
	m.Cross = (&SourceReport{}).CrossMeasurement()
	if _, err := m.Rates(t11Now, 16<<20); err == nil || !strings.Contains(err.Error(), "cross-region") {
		t.Fatal(err)
	}
	if got := (*SourceReport)(nil).CrossMeasurement(); got != (MeasuredRate{}) {
		t.Fatalf("%+v", got)
	}
}

func TestT11BudgetCheck(t *testing.T) {
	// 100 MiB database at the spec's example rates fits the default window.
	res := BudgetCheck(100, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
	if !res.Check.OK || res.AttemptsAdmitted < 1 {
		t.Fatalf("%+v", res)
	}
	if res.Rates.Cross != 10 || res.Rates.MirrorBar != 16 {
		t.Fatalf("%+v", res.Rates)
	}
	if res.Seconds <= 0 || res.Seconds > CutWindowDefault {
		t.Fatalf("%v", res.Seconds)
	}

	// A database too big for the window admits no attempt: T11 fails and
	// the reason says what was needed.
	res = BudgetCheck(100000, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now)
	if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "120") {
		t.Fatalf("%+v", res)
	}

	// A missing measurement fails closed, naming which.
	m := goodMeasurements()
	m.Disk = MeasuredRate{}
	res = BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
	if res.Check.OK || res.AttemptsAdmitted != 0 || !strings.Contains(res.Check.Reason, "no fresh rate measurement for the disk") {
		t.Fatalf("%+v", res)
	}

	// A database size that is not a number, or a window that is not, fails.
	for _, db := range []float64{math.NaN(), math.Inf(1), -1} {
		if r := BudgetCheck(db, 0, goodMeasurements(), 16<<20, CutWindowDefault, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
			t.Fatalf("db %v: %+v", db, r)
		}
	}
	for _, w := range []float64{math.NaN(), math.Inf(1), 0, -3} {
		if r := BudgetCheck(100, 0, goodMeasurements(), 16<<20, w, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 {
			t.Fatalf("window %v: %+v", w, r)
		}
	}
}

// F28(e): an override of the ready-delta bar reaches the estimate. A larger
// mirror bar costs more seconds, so a window that fits the default bar can
// refuse a bar of 4 GiB.
func TestT11BudgetCheckPricesTheOperatorsBar(t *testing.T) {
	m := goodMeasurements()
	def := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
	big := BudgetCheck(100, 0, m, 4<<30, CutWindowDefault, t11Now)
	if !def.Check.OK {
		t.Fatalf("%+v", def)
	}
	if big.Rates.MirrorBar != 4096 {
		t.Fatalf("the bar did not reach the rates: %+v", big.Rates)
	}
	if big.Seconds <= def.Seconds {
		t.Fatalf("a 4 GiB bar priced at %v, the default at %v", big.Seconds, def.Seconds)
	}
	if big.Check.OK {
		t.Fatalf("a 4 GiB mirror at 10 MiB/s fits 120 s: %+v", big)
	}
	// A bar of zero bytes is priced at the default, never below it.
	zero := BudgetCheck(100, 0, m, 0, CutWindowDefault, t11Now)
	if zero.Seconds != def.Seconds {
		t.Fatalf("%v vs %v", zero.Seconds, def.Seconds)
	}
	// A rejected budget never promotes: recording the 4 GiB result in a
	// report leaves it with zero attempts, and Ready refuses it.
	src := goodSource(t11Now)
	tr := passingTarget(t11Now)
	tr.RecordBudget(100, 0, m, big)
	if tr.AttemptsAdmitted != 0 {
		t.Fatalf("%d", tr.AttemptsAdmitted)
	}
	if ok, _ := Ready(0, 4<<30, tr, src, t11Now); ok {
		t.Fatal("a report whose budget admits no attempt promoted")
	}
}

func TestProbeKeyIsOutsideTheStagingPrefix(t *testing.T) {
	host, op := "a.go.example.com", "mv_abcdefghijklmnopqrstuvwxyz"
	k := ProbeKey(host, op)
	if k != "move-probes/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/probe" {
		t.Fatal(k)
	}
	if strings.HasPrefix(k, "moves/") || strings.HasPrefix(k, StagingPrefix(host, op)) {
		t.Fatal("the probe object is under the staging prefix")
	}
}

// ---- PutReplace ----

// casStore is an in-memory release.Store with the CAS rules of the real
// ones. beforePut runs once ahead of the first Put, so a test can lose a
// race.
type casStore struct {
	mu        sync.Mutex
	objs      map[string][]byte
	etags     map[string]int
	puts      int
	beforePut func()
	failGet   error
}

func newCASStore() *casStore {
	return &casStore{objs: map[string][]byte{}, etags: map[string]int{}}
}

func (s *casStore) etag(k string) string {
	if _, ok := s.objs[k]; !ok {
		return ""
	}
	return "e" + string(rune('0'+s.etags[k]))
}

func (s *casStore) Get(_ context.Context, key string) ([]byte, string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.failGet != nil {
		return nil, "", s.failGet
	}
	b, ok := s.objs[key]
	if !ok {
		return nil, "", fmt.Errorf("no such key %q: %w", key, os.ErrNotExist)
	}
	return append([]byte(nil), b...), s.etag(key), nil
}

func (s *casStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
	if s.beforePut != nil {
		f := s.beforePut
		s.beforePut = nil
		f()
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.puts++
	if s.etag(key) != ifMatch {
		return "", release.ErrCASConflict
	}
	s.objs[key] = append([]byte(nil), body...)
	s.etags[key]++
	return s.etag(key), nil
}

func (s *casStore) List(context.Context, string) ([]string, error)     { return nil, nil }
func (s *casStore) ListDirs(context.Context, string) ([]string, error) { return nil, nil }
func (s *casStore) Delete(context.Context, string) error               { return nil }

func TestPutReplaceCreatesThenReplaces(t *testing.T) {
	ctx := context.Background()
	s := newCASStore()
	key := TargetReportKey(ReportPrefix("bab", "app", "h", "mv_x"))
	// A create-only Put refuses the second write: the reason PutReplace exists.
	if _, err := s.Put(ctx, "other", []byte("1"), ""); err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, "other", []byte("2"), ""); !errors.Is(err, release.ErrCASConflict) {
		t.Fatal(err)
	}
	for _, body := range []string{"one", "two", "three"} {
		if err := PutReplace(ctx, s, key, []byte(body)); err != nil {
			t.Fatalf("%s: %v", body, err)
		}
		if got, _, _ := s.Get(ctx, key); string(got) != body {
			t.Fatalf("%q", got)
		}
	}
}

func TestPutReplaceRetriesOnceOnAConflict(t *testing.T) {
	ctx := context.Background()
	s := newCASStore()
	if err := PutReplace(ctx, s, "k", []byte("old")); err != nil {
		t.Fatal(err)
	}
	// Someone else replaces the object between our Get and our Put.
	s.beforePut = func() {
		s.objs["k"] = []byte("theirs")
		s.etags["k"]++
	}
	if err := PutReplace(ctx, s, "k", []byte("ours")); err != nil {
		t.Fatal(err)
	}
	if got, _, _ := s.Get(ctx, "k"); string(got) != "ours" {
		t.Fatalf("%q", got)
	}
}

func TestPutReplaceGivesUpAfterTheSecondConflict(t *testing.T) {
	ctx := context.Background()
	s := &alwaysConflict{casStore: newCASStore()}
	err := PutReplace(ctx, s, "k", []byte("x"))
	if !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("%v", err)
	}
	if s.puts != 2 {
		t.Fatalf("%d puts, want exactly one retry", s.puts)
	}
}

type alwaysConflict struct{ *casStore }

func (a *alwaysConflict) Put(_ context.Context, _ string, _ []byte, _ string) (string, error) {
	a.puts++
	return "", release.ErrCASConflict
}

func TestPutReplaceDoesNotWriteBlindOnAReadError(t *testing.T) {
	s := newCASStore()
	s.failGet = errors.New("503 slow down")
	err := PutReplace(context.Background(), s, "k", []byte("x"))
	if err == nil || !strings.Contains(err.Error(), "503") {
		t.Fatalf("%v", err)
	}
	if s.puts != 0 {
		t.Fatal("wrote after a failed read")
	}
}

func TestPutReplaceWorksOnADirStore(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	for _, body := range []string{"one", "two"} {
		if err := PutReplace(ctx, s, "moves/a/b/target.json", []byte(body)); err != nil {
			t.Fatal(err)
		}
	}
	if got, _, err := s.Get(ctx, "moves/a/b/target.json"); err != nil || string(got) != "two" {
		t.Fatalf("%q %v", got, err)
	}
}

// A report dated ahead of the reader's clock would stay fresh until the
// clock caught up and then some. One minute ahead is clock skew between
// boxes and is accepted; beyond that is refused.
func TestReadyBoundsReportTimestampsInTheFuture(t *testing.T) {
	now := t11Now
	// A target report one minute ahead is fine (its own measurements are
	// dated relative to it, so they sit in the past of now).
	tr := passingTarget(now.Add(ReportSkew))
	if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); !ok {
		t.Fatalf("a target report one minute ahead: %s", why)
	}
	tr = passingTarget(now.Add(ReportSkew + time.Nanosecond))
	if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); ok || !strings.Contains(why, "ahead") || !strings.Contains(why, "target") {
		t.Fatalf("a target report beyond the skew: %v %q", ok, why)
	}
	// Far ahead, as a stuck or wrong clock would write.
	if ok, _ := Ready(0, 16<<20, passingTarget(now.Add(24*time.Hour)), goodSource(now), now); ok {
		t.Fatal("a target report a day ahead promoted")
	}

	sr := goodSource(now)
	sr.At = now.Add(ReportSkew)
	if ok, why := Ready(0, 16<<20, passingTarget(now), sr, now); !ok {
		t.Fatalf("a source report one minute ahead: %s", why)
	}
	sr.At = now.Add(ReportSkew + time.Nanosecond)
	if ok, why := Ready(0, 16<<20, passingTarget(now), sr, now); ok || !strings.Contains(why, "ahead") || !strings.Contains(why, "source") {
		t.Fatalf("a source report beyond the skew: %v %q", ok, why)
	}
	sr.At = now.Add(24 * time.Hour)
	if ok, _ := Ready(0, 16<<20, passingTarget(now), sr, now); ok {
		t.Fatal("a source report a day ahead promoted")
	}

	// The old edge is unchanged: exactly ReportFreshness old is fresh.
	if ok, why := Ready(0, 16<<20, passingTarget(now.Add(-ReportFreshness)), goodSource(now), now); !ok {
		t.Fatal(why)
	}
	if ok, _ := Ready(0, 16<<20, passingTarget(now.Add(-ReportFreshness-time.Nanosecond)), goodSource(now), now); ok {
		t.Fatal("a report a nanosecond past the freshness window promoted")
	}
}

// The exact edges of a measurement's window: an hour old and a minute
// ahead are accepted, a nanosecond beyond either is refused.
func TestMeasurementWindowEdges(t *testing.T) {
	at := func(d time.Duration) error {
		m := goodMeasurements()
		m.S3.At = t11Now.Add(d)
		_, err := m.Rates(t11Now, 16<<20)
		return err
	}
	for _, c := range []struct {
		d  time.Duration
		ok bool
	}{
		{-MeasurementFreshness, true},
		{-MeasurementFreshness - time.Nanosecond, false},
		{MeasurementSkew, true},
		{MeasurementSkew + time.Nanosecond, false},
		{0, true},
	} {
		if err := at(c.d); (err == nil) != c.ok {
			t.Errorf("age %v: err %v, want ok=%v", c.d, err, c.ok)
		}
	}
}

// Ruling T7-budget: the archive's entry count is priced (and carried in the
// report, so Ready re-prices it); a count that is not usable fails closed.
func TestT11BudgetCheckPricesEntries(t *testing.T) {
	m := goodMeasurements()
	none := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now)
	many := BudgetCheck(100, 50000, m, 16<<20, CutWindowDefault, t11Now)
	if !none.Check.OK || many.Seconds <= none.Seconds {
		t.Fatalf("entries were not priced: %v vs %v", many.Seconds, none.Seconds)
	}
	if r := BudgetCheck(100, -1, m, 16<<20, CutWindowDefault, t11Now); r.Check.OK || r.AttemptsAdmitted != 0 || r.Check.Reason != "the archive's file count could not be measured" {
		t.Fatalf("%+v", r)
	}
	tr := passingTarget(t11Now)
	tr.RecordBudget(100, 50000, m, many)
	if tr.Entries != 50000 || tr.Rates.FsyncRate != 1600 || !tr.FsyncAt.Equal(m.Fsync.At) {
		t.Fatalf("RecordBudget dropped the entry evidence: %+v", tr)
	}
}

// The release a promotion pins is one the target checked (T7): a target
// report with no release, or one for another release than the source now
// runs, never promotes (Task 14 review finding 3).
func TestReadyNeedsTheReleaseTheTargetChecked(t *testing.T) {
	now := t11Now
	if ok, why := Ready(0, 16<<20, passingTarget(now), goodSource(now), now); !ok {
		t.Fatal(why)
	}
	tr := passingTarget(now)
	tr.Inputs.Release = ""
	if ok, why := Ready(0, 16<<20, tr, goodSource(now), now); ok || why != "the target has not checked a release yet" {
		t.Fatalf("no release checked: %v %q", ok, why)
	}
	src := goodSource(now)
	src.Release = "v8"
	if ok, why := Ready(0, 16<<20, passingTarget(now), src, now); ok || why != "the target checked release v7 and the source now runs v8" {
		t.Fatalf("release drift: %v %q", ok, why)
	}
	src.Release = ""
	tr = passingTarget(now)
	tr.Inputs.Release = ""
	if ok, _ := Ready(0, 16<<20, tr, src, now); ok {
		t.Fatal("two empty releases promoted")
	}
}

// Measurements.Rates names the stale rate as data (*NoFreshRateError), and
// BudgetCheck builds its reason from that, never from an error's text.
func TestNoFreshRateIsTyped(t *testing.T) {
	m := goodMeasurements()
	m.S3 = MeasuredRate{}
	_, err := m.Rates(t11Now, 16<<20)
	var nf *NoFreshRateError
	if !errors.As(err, &nf) || nf.Which != "regional S3" {
		t.Fatalf("%v", err)
	}
	if r := BudgetCheck(100, 0, m, 16<<20, CutWindowDefault, t11Now); r.Check.Reason != NoFreshRate("regional S3") {
		t.Fatalf("%q", r.Check.Reason)
	}
}

// Final review finding 4: an archive reference pins its objects only by
// version ids that name one immutable version each: neither "" nor the
// suspended bucket's "null".
func TestArchiveRefPinnedNeedsImmutableVersions(t *testing.T) {
	good := ArchiveRef{ManifestKey: "m", ManifestVersion: "mv1", SealedObjectKey: "s", SealedVersion: "sv1"}
	if !good.Pinned() {
		t.Fatal("a reference with real version ids is not pinned")
	}
	for _, c := range []struct{ manifest, sealed string }{{"null", "sv1"}, {"mv1", "null"}, {"", "sv1"}, {"mv1", ""}, {"null", "null"}} {
		r := good
		r.ManifestVersion, r.SealedVersion = c.manifest, c.sealed
		if r.Pinned() {
			t.Errorf("versions %q and %q read as pinned", c.manifest, c.sealed)
		}
	}
}

// Final review finding 2: the target's report priced the cut (T11) and
// checked the disk (T6) for the database size it saw. When the source's
// report now carries a larger one (its local file, or the logical size of
// its replica's newest LTX), the target's evidence no longer covers it and
// Ready refuses until the target reports again. The review's example: a
// fresh report for 100 MiB, then the database grows to 2.5 GiB, which at
// 10/250/500 MiB/s no longer fits the 120-second window.
func TestReadyRefusesADatabaseGrownSinceTheTargetPriced(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	target := passingTarget(now.Add(-time.Minute))
	m := Measurements{
		Cross: MeasuredRate{MiBPerS: 10, At: now.Add(-time.Minute)},
		S3:    MeasuredRate{MiBPerS: 250, At: now.Add(-time.Minute)},
		Disk:  MeasuredRate{MiBPerS: 500, At: now.Add(-time.Minute)},
		Fsync: FilesRate{PerS: 1600, At: now.Add(-time.Minute)},
	}
	if res := BudgetCheck(2560, 0, m, 16<<20, 120, now); res.Check.OK {
		t.Fatalf("the example does not overrun its window: %+v", res)
	}
	const grown = 2560 << 20
	for name, src := range map[string]func(*SourceReport){
		"local file": func(s *SourceReport) { s.DBBytes = grown },
		"replica":    func(s *SourceReport) { s.ReplicaBytes = grown },
		"one byte":   func(s *SourceReport) { s.ReplicaBytes = 100<<20 + 1 },
	} {
		s := goodSource(now)
		src(s)
		ok, why := Ready(0, 16<<20, target, s, now)
		if ok || !strings.HasPrefix(why, "T11: the database has grown to ") {
			t.Errorf("%s: Ready = %v %q, want the grown-database refusal", name, ok, why)
		}
	}
	want := "T11: the database has grown to 2560 MiB since the target priced 100 MiB. waiting for the target's next report"
	s := goodSource(now)
	s.DBBytes = grown
	if _, why := Ready(0, 16<<20, target, s, now); why != want {
		t.Errorf("reason %q, want %q", why, want)
	}
	// The size the target priced, or less, is covered.
	for _, size := range []int64{0, 1, 100 << 20} {
		s := goodSource(now)
		s.DBBytes, s.ReplicaBytes = size, size
		if ok, why := Ready(0, 16<<20, target, s, now); !ok {
			t.Errorf("size %d: not ready: %s", size, why)
		}
	}
}

// The cut guard (Paul's ruling, 2026-10-11): T13 is the target's statement
// that it can land the move and roll it back. Only a report that states it
// passing is armed: a nil report, an older box's report that has no T13,
// and a failing T13 all read unarmed, and an older report is not ready.
func TestTargetReportArmedFailsClosed(t *testing.T) {
	now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
	var none *TargetReport
	if none.Armed() {
		t.Fatal("a missing report is armed")
	}
	r := passingTarget(now)
	if !r.Armed() {
		t.Fatal("a report with a passing T13 is not armed")
	}
	r.Checks[CheckArmed] = Check{Reason: "not armed"}
	if r.Armed() {
		t.Fatal("a failing T13 is armed")
	}
	old := passingTarget(now)
	delete(old.Checks, CheckArmed)
	if old.Armed() {
		t.Fatal("an older box's report, with no T13, is armed")
	}
	if ok, why := Ready(0, 16<<20, old, goodSource(now), now); ok || why != "T13 was not reported" {
		t.Fatalf("an older box's report: ready %v (%q)", ok, why)
	}
	// The wire form: an older box's JSON has no T13 and reads unarmed.
	body, err := json.Marshal(old)
	if err != nil {
		t.Fatal(err)
	}
	var back TargetReport
	if err := json.Unmarshal(body, &back); err != nil || back.Armed() {
		t.Fatalf("round trip of an older report: armed %v, %v", back.Armed(), err)
	}
}