package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// The source's finalization harness: the landing harness driven through
// restored, then committed the way Task 7 will (the record's probing and
// probed steps, MoveCommit, and hostclaim.Commit with the source's location
// park scoped by sourceLocationParkScope), and the SOURCE box (ie-box) given
// a real registry holding its row "moving" under the move with two timers,
// the config ack the box last wrote, and the database's marker and
// quarantine beside the database the cut harness left at the flat path.
// The source's park engine runs over the fake versioned store and the same
// fake lease table as the landing's (fakeParkLeases).

const finalAckBody = `{"v":1,"host":"a.go.example.com"}`

type finalHarness struct {
	*landingHarness
	reg       *registry.Registry
	finalAt   string // the source's crash point; "" crashes nowhere
	finalDone bool
	finalHits []string
	crashAll  map[string]bool // crash once at every point not in it (the replay test)
	deletes   int             // DeleteHeld calls that deleted
}

func newFinalHarness(t *testing.T, opts ...p1Option) *finalHarness {
	t.Helper()
	return newFinalHarnessFrom(t, newLandingHarness(t, opts...))
}

// newFinalHarnessFrom is newFinalHarness over a landing harness the caller
// prepared before the flip.
func newFinalHarnessFrom(t *testing.T, lh *landingHarness) *finalHarness {
	t.Helper()
	lh.flip(t)
	return newFinalHarnessFlipped(t, lh)
}

// newFinalHarnessFlipped is newFinalHarnessFrom over a landing harness the
// caller already flipped (and may have changed after the flip).
func newFinalHarnessFlipped(t *testing.T, lh *landingHarness) *finalHarness {
	t.Helper()
	h := &finalHarness{landingHarness: lh}
	h.land(t)
	h.installed(t)
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	h.reg = reg
	src := h.phase1Harness.route
	src.Status, src.HoldOp = "moving", p1ID
	if err := reg.Upsert(src); err != nil {
		t.Fatal(err)
	}
	for _, tm := range []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000},
		{Host: p1Host, Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000}} {
		if err := reg.UpsertTimer(tm); err != nil {
			t.Fatal(err)
		}
	}
	dd := h.cfg.DataDir
	for name, body := range map[string]string{p1Host + ".db.gen": "cur", p1Host + ".db.corrupt": "quarantined",
		p1Host + ".config.env": "SECRET=1\n", p1Host + "-sidecar.env": "SIDE=1\n"} {
		if err := os.WriteFile(filepath.Join(dd, name), []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := h.s.Put(context.Background(), appconfig.ConfigAckKey(h.srcBox, p1Host), []byte(finalAckBody), ""); err != nil {
		t.Fatal(err)
	}
	h.cfg.Route = func(host string) (*registry.Route, error) { return reg.Get(host) }
	h.cfg.OwnsRoute = func(host string) bool {
		rt, err := reg.Get(host)
		return err == nil && rt != nil && rt.Provisioned
	}
	h.cfg.DeleteHeld = func(host, op string) (bool, error) {
		ok, err := reg.DeleteIfHoldWithTimers(host, op)
		if ok {
			h.deletes++
		}
		return ok, err
	}
	h.cfg.Guard = hostguard.Guard{DataDir: dd, LockDir: t.TempDir()}
	h.cfg.Local = park.Local{DataDir: dd, Box: h.srcBox}
	h.cfg.ParkLeases = h.leases
	h.cfg.ParkView = sourceTestViewFor(park.PlacementRegion(h.src), park.PlacementRegion(h.dst))
	h.cfg.ParkAuthorize = sourceTestAuthorizeFor(h.srcBox, park.PlacementRegion(h.src))
	h.cfg.ParkLocalAuthorize = sourceTestLocalAuthorizeFor(h.srcBox, park.PlacementRegion(h.src))
	h.cfg.parkFault = h.lcfg.parkFault
	h.cfg.finalFault = func(point string) error {
		h.finalHits = append(h.finalHits, point)
		if h.crashAll != nil && !h.crashAll[point] {
			h.crashAll[point] = true
			return errCrash
		}
		if point == h.finalAt && !h.finalDone {
			h.finalDone = true
			return errCrash
		}
		return nil
	}
	return h
}

// advanceToProbed walks the record and the claim from restored through
// probing to probed, as Task 6 will.
func (h *finalHarness) advanceToProbed(t *testing.T) {
	t.Helper()
	ctx := context.Background()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	for _, step := range [][2]string{{provision.LandRestored, provision.LandProbing}, {provision.LandProbing, provision.LandProbed}} {
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveLandingPhase(g, step[0], step[1], nil)); err != nil {
			t.Fatal(err)
		}
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.SetMovePhase(p1ID, hostclaim.KindLanding, step[0], step[1])); err != nil {
			t.Fatal(err)
		}
	}
}

// commitRecord is the record's commit alone (MoveCommit).
func (h *finalHarness) commitRecord(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveCommit(g, h.clock.Now())); err != nil {
		t.Fatal(err)
	}
}

// commitClaim is the claim's commit (hostclaim.Commit), with the source's
// location park scoped from the flip (H7) and the target's staging lane.
func (h *finalHarness) commitClaim(t *testing.T) {
	t.Helper()
	ctx := context.Background()
	rec := h.read(t, p1Host)
	obs, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	scope, err := sourceLocationParkScope(rec, *obs.Claim.Holder)
	if err != nil {
		t.Fatal(err)
	}
	staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
	if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.Commit(p1ID, park.PlacementRegion(h.src), h.srcBox, park.PlacementRegion(h.dst), scope, staging, obligationID)); err != nil {
		t.Fatal(err)
	}
}

func (h *finalHarness) commit(t *testing.T) {
	t.Helper()
	h.advanceToProbed(t)
	h.commitRecord(t)
	h.commitClaim(t)
}

// sourcePass runs one pass as the source box.
func (h *finalHarness) sourcePass(t *testing.T) { t.Helper(); h.pass(t) }

// crashSource runs one source pass that must reach the crash point.
func (h *finalHarness) crashSource(t *testing.T) {
	t.Helper()
	h.sourcePass(t)
	if !h.finalDone {
		t.Fatalf("the pass never reached %s (points reached: %v)", h.finalAt, h.finalHits)
	}
}

func (h *finalHarness) sourceJournal(t *testing.T) *moves.Journal {
	t.Helper()
	j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
	if err != nil {
		t.Fatal(err)
	}
	return j
}

// live lists which of the source's artifacts are at their live paths.
func (h *finalHarness) live(t *testing.T) []string {
	t.Helper()
	var out []string
	for _, p := range append(testSourcePaths(), filepath.Join(".moves", p1Host, p1ID)) {
		if _, err := os.Lstat(filepath.Join(h.cfg.DataDir, p)); err == nil {
			out = append(out, p)
		} else if !errors.Is(err, os.ErrNotExist) {
			t.Fatal(err)
		}
	}
	return out
}

// parked lists every regular file under .parked/<inc>/<host>/, as
// "<retirement>/<rest>", sorted.
func (h *finalHarness) parked(t *testing.T) []string {
	t.Helper()
	root := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host)
	var out []string
	filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
		if err == nil && !d.IsDir() {
			rel, _ := filepath.Rel(root, p)
			out = append(out, filepath.ToSlash(rel))
		}
		return nil
	})
	sort.Strings(out)
	return out
}

// untouched asserts that nothing of the source's was finalized.
func (h *finalHarness) untouched(t *testing.T, why string) {
	t.Helper()
	if got, want := h.live(t), []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}; strings.Join(got, " ") != strings.Join(want, " ") {
		t.Fatalf("%s: live artifacts %v, want %v", why, got, want)
	}
	if got := h.parked(t); len(got) != 0 {
		t.Fatalf("%s: parked %v", why, got)
	}
	if rt, err := h.reg.Get(p1Host); err != nil || rt == nil {
		t.Fatalf("%s: the row: %+v %v", why, rt, err)
	}
	if tm, _ := h.reg.TimersFor(p1Host); len(tm) != 2 {
		t.Fatalf("%s: timers %+v", why, tm)
	}
	if m := h.read(t, p1Host).Move; m == nil || m.SourceParkedAt != nil {
		t.Fatalf("%s: record %+v", why, m)
	}
	if _, _, err := h.s.Get(context.Background(), appconfig.ConfigAckKey(h.srcBox, p1Host)); err != nil {
		t.Fatalf("%s: the config ack: %v", why, err)
	}
	for _, step := range []string{finalStepRemote, finalStepInstallation, finalStepLocal + "-plan", finalStepRowDeleted} {
		if h.sourceJournal(t).Done(step) {
			t.Fatalf("%s: the journal has %s", why, step)
		}
	}
	if calls := h.vs.callsOn("parked/"); len(calls) != 0 {
		t.Fatalf("%s: the source's location was parked: %v", why, calls)
	}
}

// finalized asserts the end state of a source finalization of an
// unregistered source: every artifact in one retirement under the engine's
// layout with the op's journal and a receipt naming the op, nothing live,
// the row and its timers gone, the config ack moved into the report
// prefix, source_parked_at stamped, and the source's location parked.
func (h *finalHarness) finalized(t *testing.T) string {
	t.Helper()
	ctx := context.Background()
	if got := h.live(t); len(got) != 0 {
		t.Fatalf("still live: %v", got)
	}
	parked := h.parked(t)
	if len(parked) == 0 {
		t.Fatal("nothing parked")
	}
	ret, _, _ := strings.Cut(parked[0], "/")
	var rest []string
	for _, p := range parked {
		r, tail, _ := strings.Cut(p, "/")
		if r != ret {
			t.Fatalf("more than one retirement: %v", parked)
		}
		rest = append(rest, tail)
	}
	jpath := "artifacts/.moves/" + p1Host + "/" + p1ID + "/journal.json"
	want := []string{"artifacts/" + p1Host + "/upload.txt", "artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.corrupt", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env",
		"artifacts/" + p1Host + ".config.env", "artifacts/" + p1Host + "-sidecar.env", localParkReceiptName}
	var got []string
	sawJournal := false
	for _, r := range rest {
		switch {
		case r == jpath:
			sawJournal = true
		case strings.HasPrefix(r, "artifacts/.moves/"):
			// the op journal directory's other contents (the archive scratch)
		default:
			got = append(got, r)
		}
	}
	sort.Strings(want)
	if strings.Join(got, " ") != strings.Join(want, " ") || !sawJournal {
		t.Fatalf("parked %v (journal %v), want %v and the journal", got, sawJournal, want)
	}
	art := filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, "artifacts")
	for name, body := range map[string]string{p1Host + ".db": "the stopped database", p1Host + "/upload.txt": "hello", p1Host + ".env": "APP_MODE=prod\n", p1Host + ".db.gen": "cur"} {
		if b, err := os.ReadFile(filepath.Join(art, name)); err != nil || string(b) != body {
			t.Fatalf("parked %s: %q %v", name, b, err)
		}
	}
	var receipt localParkReceipt
	if b, err := os.ReadFile(filepath.Join(filepath.Dir(art), localParkReceiptName)); err != nil || json.Unmarshal(b, &receipt) != nil || receipt.Op != p1ID || receipt.Retirement != ret || receipt.Incarnation != h.inc {
		t.Fatalf("receipt %+v %v", receipt, err)
	}
	var pj moves.Journal
	if b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, ret, jpath)); err != nil || json.Unmarshal(b, &pj) != nil {
		t.Fatalf("the parked journal: %v", err)
	}
	if pj.Steps[finalStepRowDeleted] != "done" || !pj.Done(finalStepParked) || pj.Steps[finalStepLocal] != ret || pj.Steps[finalStepInstallation] != "unregistered" {
		t.Fatalf("the parked journal's steps: %v", pj.Steps)
	}
	if rt, err := h.reg.Get(p1Host); err != nil || rt != nil {
		t.Fatalf("the row: %+v %v", rt, err)
	}
	if tm, err := h.reg.TimersFor(p1Host); err != nil || len(tm) != 0 {
		t.Fatalf("timers: %+v %v", tm, err)
	}
	if _, _, err := h.s.Get(ctx, appconfig.ConfigAckKey("ie-box", p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the config ack is still there: %v", err)
	}
	if b, _, err := h.s.Get(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID)+finalConfigAckName); err != nil || string(b) != finalAckBody {
		t.Fatalf("the config ack's copy: %q %v", b, err)
	}
	m := h.read(t, p1Host).Move
	if m == nil || m.State != provision.MoveCommitted || m.SourceParkedAt == nil {
		t.Fatalf("record: %+v", m)
	}
	// The source's location is archived and gone from its live keys; its
	// park-replica is still owed until the horizon (the regional runner's).
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, cutRoot+"/") {
			t.Fatalf("still at the source's location: %s", k)
		}
	}
	archived := 0
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") && strings.Contains(k, "/instances/") {
			archived++
		}
	}
	if archived == 0 {
		t.Fatal("nothing archived from the source's location")
	}
	if held := h.leases.holding(); len(held) != 0 {
		t.Fatalf("leases still held: %v", held)
	}
	return ret
}

// Review Focus 5: the source finalizes only a committed record whose row
// this move still holds (or whose journal says the row is deleted).
func TestSourceParkWaitsForCommit(t *testing.T) {
	t.Parallel()
	t.Run("a landing record", func(t *testing.T) {
		h := newFinalHarness(t)
		h.advanceToProbed(t)
		h.sourcePass(t)
		h.untouched(t, "landing/probed")
	})
	t.Run("a committed record whose claim is not committed", func(t *testing.T) {
		h := newFinalHarness(t)
		h.advanceToProbed(t)
		h.commitRecord(t)
		h.sourcePass(t)
		h.untouched(t, "claim at probed")
	})
	t.Run("a committed record whose row another op holds", func(t *testing.T) {
		h := newFinalHarness(t)
		h.commit(t)
		if ok, err := h.reg.ReleaseHold(p1Host, p1ID); err != nil || !ok {
			t.Fatal(ok, err)
		}
		if err := h.reg.SetHold(p1Host, "restoring", p1ID2); err != nil {
			t.Fatal(err)
		}
		h.sourcePass(t)
		h.untouched(t, "row held by another op")
		if rt, _ := h.reg.Get(p1Host); rt.HoldOp != p1ID2 {
			t.Fatalf("the other op's row: %+v", rt)
		}
	})
	t.Run("a committed record with no row and no journalled delete", func(t *testing.T) {
		h := newFinalHarness(t)
		h.commit(t)
		if err := h.reg.Delete(p1Host); err != nil {
			t.Fatal(err)
		}
		h.sourcePass(t)
		if got := h.live(t); len(got) != 8 || len(h.parked(t)) != 0 || h.read(t, p1Host).Move.SourceParkedAt != nil {
			t.Fatalf("finalized without the row: live %v parked %v", got, h.parked(t))
		}
	})
	t.Run("then the commit", func(t *testing.T) {
		h := newFinalHarness(t)
		h.advanceToProbed(t)
		h.sourcePass(t)
		h.commitRecord(t)
		h.commitClaim(t)
		h.sourcePass(t)
		h.finalized(t)
	})
}

// The end to end finalization of an unregistered source (ruling H5(b)): the
// database, its marker and quarantine, the files directory and the env are
// gone from the live paths and in one retirement under the engine's layout;
// the row and its timers rows are gone; source_parked_at is set; the op
// journal is parked last; a further pass changes nothing.
func TestSourceParkRetiresInstallationAndRow(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	if c := h.claim(t); len(c.Obligations) != 2 {
		t.Fatalf("the commit queued %+v (want the source's location and the target's lane, no park-local)", c.Obligations)
	}
	h.sourcePass(t)
	h.finalized(t)
	if h.deletes != 1 {
		t.Fatalf("the row was deleted %d times", h.deletes)
	}
	before := h.parked(t)
	h.freshWorker()
	h.sourcePass(t)
	if after := h.parked(t); strings.Join(after, " ") != strings.Join(before, " ") || h.deletes != 1 {
		t.Fatalf("a further pass changed the park: %v", after)
	}
	if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
		t.Fatal("a finalized source still routes the record here")
	}
}

// Past the horizon the engine acknowledges the source's location park in
// the same run, and the finalization goes on.
func TestSourceParkPastTheHorizonAcknowledges(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	h.clock.Advance(park.SignatureAge + cutRequestLimit + time.Minute)
	h.sourcePass(t)
	h.finalized(t)
	for _, o := range h.claim(t).Obligations {
		if o.Kind == "park-replica" && o.Region == "ie" {
			t.Fatalf("the source's location park is still owed: %+v", o)
		}
	}
}

// The remote half comes first: while the archive is incomplete (a delete
// the store accepts but does not land) nothing local moves.
func TestSourceParkWaitsForTheRemoteHalf(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	stuck := true
	h.vs.skipDelete = func(key string) bool { return stuck && strings.HasPrefix(key, cutRoot+"/") }
	h.sourcePass(t)
	if got := h.live(t); len(got) != 8 {
		t.Fatalf("the local half ran before the remote: live %v", got)
	}
	if j := h.sourceJournal(t); j.Done(finalStepRemote) || j.Done(finalStepLocal+"-plan") {
		t.Fatalf("journal: %v", j.Steps)
	}
	if rt, _ := h.reg.Get(p1Host); rt == nil {
		t.Fatal("the row went before the remote half")
	}
	stuck = false
	h.sourcePass(t)
	h.finalized(t)
}

// Round-2 finding 12: a crash between the row's delete and the journal's
// "done" (and before MoveSourceParked) is replayed: the record's region is
// the target's and the row is gone, so only the op journal routes the
// record back here, and the replay treats the pending delete as done and
// stamps source_parked_at.
func TestSourceParkSurvivesACrashAfterTheRowDelete(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"after-row-pending", "after-row-delete"} {
		t.Run(point, func(t *testing.T) {
			h := newFinalHarness(t)
			h.commit(t)
			h.finalAt = point
			h.crashSource(t)
			if j := h.sourceJournal(t); j.Steps[finalStepRowDeleted] != "pending" {
				t.Fatalf("journal: %v", j.Steps)
			}
			rt, _ := h.reg.Get(p1Host)
			if (point == "after-row-delete") != (rt == nil) {
				t.Fatalf("the row after %s: %+v", point, rt)
			}
			if h.read(t, p1Host).Move.SourceParkedAt != nil {
				t.Fatal("stamped before the crash point")
			}
			if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleSource {
				t.Fatal("the journal does not route the record back to the source")
			}
			h.freshWorker()
			h.sourcePass(t)
			h.finalized(t)
			if h.deletes != 1 {
				t.Fatalf("deleted %d times", h.deletes)
			}
		})
	}
}

// Every crash point, each once, with a fresh worker after each: the replay
// converges on the same end state, with each artifact renamed once, the row
// deleted once and one retirement.
func TestSourceParkReplayIsIdempotent(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	h.crashAll = map[string]bool{}
	h.parkAt = "delete"
	for i := 0; i < 40 && len(h.live(t)) > 0; i++ {
		h.freshWorker()
		h.sourcePass(t)
	}
	h.finalized(t)
	if h.deletes != 1 {
		t.Fatalf("deleted %d times", h.deletes)
	}
	want := []string{"after-" + finalStepRemote, "after-" + finalStepInstallation, "after-" + finalStepLocal + "-plan",
		"after-" + finalStepLocal,
		"after-row-pending", "after-row-delete", "after-" + finalStepRowDeleted, "after-config-ack-copy", "after-" + finalStepConfigAck,
		"after-source-parked-pending", "after-source-parked-record", "after-" + finalStepParked, "after-" + finalStepJournal}
	for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env"} {
		want = append(want, "after-rename-unsynced:"+p, "after-rename:"+p)
	}
	for _, p := range want {
		if !h.crashAll[p] {
			t.Errorf("never crashed at %s (crashed at %v)", p, h.crashAll)
		}
	}
	if !h.parkDone {
		t.Error("the park engine never crashed at its delete")
	}
}

// A registered source (the commit queued its park-local because the
// inventory lists paths of the journalled stamp S) retires its installation
// through park.Local under the guard, which completes the park-local; what
// the inventory does not list is parked by the moves-owned rename.
func TestSourceParkRetiresARegisteredInstallation(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	dd := h.cfg.DataDir
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(dd, d), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	rec := `{"v":1,"host":"` + p1Host + `","incarnation":"` + h.inc + `","active_stamp":1,"retired_stamps":[]}`
	if err := os.WriteFile(filepath.Join(dd, "hostguard", p1Host+".json"), []byte(rec), 0o660); err != nil {
		t.Fatal(err)
	}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		if c.Op.Journal[hostclaim.JournalSourceStamp] != "1" {
			t.Fatalf("fixture: source stamp %q", c.Op.Journal[hostclaim.JournalSourceStamp])
		}
		if c.Inventory == nil {
			c.Inventory = map[string]hostclaim.InventoryEntry{}
		}
		c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}
	})
	h.commit(t)
	var chore hostclaim.Obligation
	for _, o := range h.claim(t).Obligations {
		if o.Kind == "park-local" {
			chore = o
		}
	}
	if chore.Box != "ie-box" {
		t.Fatalf("the commit queued no park-local for the source: %+v", h.claim(t).Obligations)
	}
	h.sourcePass(t)
	c := h.claim(t)
	for _, o := range c.Obligations {
		if o.Kind == "park-local" {
			t.Fatalf("the park-local is still owed: %+v", o)
		}
	}
	if len(c.Inventory["ie-box"].Paths) != 0 {
		t.Fatalf("inventory: %+v", c.Inventory["ie-box"])
	}
	if got := h.live(t); len(got) != 0 {
		t.Fatalf("still live: %v", got)
	}
	parked := strings.Join(h.parked(t), " ")
	for _, p := range []string{chore.Retirement + "/artifacts/" + p1Host + ".db", chore.Retirement + "/artifacts/" + p1Host + "/upload.txt", chore.Retirement + "/.journal.json"} {
		if !strings.Contains(parked, p) {
			t.Fatalf("park.Local's retirement lacks %s: %s", p, parked)
		}
	}
	j := h.parkedJournal(t)
	if j.Steps[finalStepInstallation] != "retired" {
		t.Fatalf("installation step %q", j.Steps[finalStepInstallation])
	}
	ret := j.Steps[finalStepLocal]
	if ret == chore.Retirement || !strings.Contains(parked, ret+"/artifacts/"+p1Host+".db.corrupt") {
		t.Fatalf("the unregistered quarantine is not in the moves-owned retirement %s: %s", ret, parked)
	}
	if captured, err := hostguard.Capture(dd, p1Host); err != nil || captured.Stamp != 0 {
		t.Fatalf("the installation record is still active: %+v %v", captured, err)
	}
	if rt, _ := h.reg.Get(p1Host); rt != nil || h.read(t, p1Host).Move.SourceParkedAt == nil {
		t.Fatal("the row or the stamp")
	}
}

// parkedJournal reads the op journal from wherever the finalization parked it.
func (h *finalHarness) parkedJournal(t *testing.T) moves.Journal {
	t.Helper()
	var j moves.Journal
	for _, p := range h.parked(t) {
		if strings.HasSuffix(p, "/.moves/"+p1Host+"/"+p1ID+"/journal.json") {
			b, err := os.ReadFile(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, p))
			if err != nil || json.Unmarshal(b, &j) != nil {
				t.Fatalf("parked journal: %v", err)
			}
			return j
		}
	}
	t.Fatalf("no parked journal in %v", h.parked(t))
	return j
}

// The rename moves only what its plan recorded: after the plan, an
// artifact replaced at its live path (another identity) and one that
// appears where the plan found nothing are both left where they are,
// untouched; everything recorded is parked.
func TestSourceParkLeavesWhatItDidNotRecord(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	dd := h.cfg.DataDir
	if err := os.Remove(filepath.Join(dd, p1Host+".db.corrupt")); err != nil {
		t.Fatal(err)
	}
	h.finalAt = "after-" + finalStepLocal + "-plan"
	h.crashSource(t)
	envPath := filepath.Join(dd, p1Host+".env")
	if err := os.Rename(envPath, envPath+".old"); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(envPath, []byte("SOMEONE=else\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dd, p1Host+".db.corrupt"), []byte("new"), 0o600); err != nil {
		t.Fatal(err)
	}
	h.freshWorker()
	h.sourcePass(t)
	for name, body := range map[string]string{p1Host + ".env": "SOMEONE=else\n", p1Host + ".db.corrupt": "new"} {
		if b, err := os.ReadFile(filepath.Join(dd, name)); err != nil || string(b) != body {
			t.Fatalf("%s was touched: %q %v", name, b, err)
		}
	}
	parked := strings.Join(h.parked(t), " ")
	if strings.Contains(parked, p1Host+".env") || strings.Contains(parked, p1Host+".db.corrupt") || !strings.Contains(parked, "artifacts/"+p1Host+".db ") {
		t.Fatalf("parked %s", parked)
	}
	if h.read(t, p1Host).Move.SourceParkedAt == nil {
		t.Fatal("the finalization did not complete")
	}
}

// H7: the source's location park's horizon starts at the flip, so a final
// checkpoint request signed after the stop evidence was collected, and
// before the flip, is inside it; a park scoped from the stop evidence is
// refused before the engine runs.
func TestSourceLocationParkScopeCoversTheFinalCheckpoint(t *testing.T) {
	t.Parallel()
	checked := time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)
	flipped := checked.Add(40 * time.Second)
	signed := checked.Add(30 * time.Second) // the final checkpoint, after CheckedAt
	rec := provision.Record{Host: p1Host, Move: &provision.MoveState{ID: p1ID, FlippedAt: &flipped}}
	scope, err := sourceLocationParkScope(rec, hostclaim.Holder{Location: cutRoot + "/", Incarnation: "inc"})
	if err != nil || scope.Location != cutRoot || !scope.EvidenceAt.Equal(flipped) || scope.RequestLimit != cutRequestLimit {
		t.Fatalf("scope %+v %v", scope, err)
	}
	horizon := scope.EvidenceAt.Add(park.SignatureAge + scope.RequestLimit)
	if latest := signed.Add(park.SignatureAge + cutRequestLimit); latest.After(horizon) {
		t.Fatalf("a request signed at %s can land until %s, after the horizon %s", signed, latest, horizon)
	}
	if early := checked.Add(park.SignatureAge + cutRequestLimit); !signed.Add(park.SignatureAge + cutRequestLimit).After(early) {
		t.Fatal("fixture: the stop evidence's horizon would have covered it too")
	}
	if _, err := sourceLocationParkScope(provision.Record{Move: &provision.MoveState{}}, hostclaim.Holder{}); err == nil {
		t.Fatal("a scope without the flip")
	}

	h := newFinalHarness(t)
	h.advanceToProbed(t)
	h.commitRecord(t)
	rec = h.read(t, p1Host)
	obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	early := hostclaim.ParkScope{Location: cutRoot, Incarnation: h.inc, EvidenceAt: rec.Move.FlippedAt.Add(-time.Minute), RequestLimit: cutRequestLimit}
	staging := hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, p1ID), Incarnation: h.inc, EvidenceAt: *rec.Move.FlippedAt, RequestLimit: cutRequestLimit}
	if _, err := hostclaim.Update(context.Background(), h.s, obs, hostclaim.Commit(p1ID, "ie", "ie-box", "uk", early, staging, obligationID)); err != nil {
		t.Fatal(err)
	}
	h.sourcePass(t)
	h.untouched(t, "a horizon before the flip")
}

// The routing rule for the states after the flip (H5(a)): this box's
// region names the role, and the row or the op's journal says it has work.
func TestMovesParticipantRecord(t *testing.T) {
	dd := t.TempDir()
	owns := false
	cfg := moveCfg{Region: "ie", DataDir: dd, OwnsRoute: func(string) bool { return owns }}
	rec := func(state, from, to string) provision.Record {
		return provision.Record{Host: p1Host, Region: to, Move: &provision.MoveState{ID: p1ID, State: state, From: from, To: to}}
	}
	if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleNone {
		t.Fatalf("no row and no journal: %q", got)
	}
	owns = true
	for _, c := range []struct {
		rec  provision.Record
		want string
	}{
		{rec(provision.MoveCommitted, "ie", "uk"), roleSource},
		{rec(provision.MoveRollingBack, "ie", "uk"), roleSource},
		{rec(provision.MoveReverting, "ie", "uk"), roleSource},
		{rec(provision.MoveCommitted, "uk", "ie"), roleTarget},
		{rec(provision.MoveCommitted, "fr", "uk"), roleNone},
		{rec(provision.MoveLanding, "ie", "uk"), roleNone},
		{rec(provision.MovePreparing, "ie", "uk"), roleNone},
		{rec(provision.MoveCommitted, "ie", "ie"), roleNone},
		{provision.Record{Host: p1Host, Move: &provision.MoveState{State: provision.MoveCommitted, From: "ie", To: "uk"}}, roleNone},
		{provision.Record{Host: p1Host}, roleNone},
	} {
		if got := movesParticipantRecord(cfg, c.rec); got != c.want {
			t.Errorf("%+v: %q, want %q", c.rec.Move, got, c.want)
		}
	}
	owns = false
	j := &moves.Journal{Host: p1Host, Op: p1ID}
	if err := j.Commit(dd); err != nil {
		t.Fatal(err)
	}
	if got := movesParticipantRecord(cfg, rec(provision.MoveCommitted, "ie", "uk")); got != roleSource {
		t.Fatalf("the op's journal: %q", got)
	}
	if got := movesParticipantRecord(cfg, provision.Record{Host: p1Host, Move: &provision.MoveState{ID: "../x", State: provision.MoveCommitted, From: "ie", To: "uk"}}); got != roleNone {
		t.Fatalf("an unusable op id: %q", got)
	}
}

// Production leaves the finalization unarmed (no park engine leases until
// Plan 5, ruling G17) but wires the registry's delete; an unarmed pass
// changes nothing.
func TestSourceParkWaitsUnarmedInProduction(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	for name, unarm := range map[string]func(*moveCfg){
		"no park leases":            func(c *moveCfg) { c.ParkLeases = nil },
		"no park view":              func(c *moveCfg) { c.ParkView = nil },
		"no park authorizer":        func(c *moveCfg) { c.ParkAuthorize = nil },
		"no claim local authorizer": func(c *moveCfg) { c.ParkLocalAuthorize = nil },
	} {
		saved := h.cfg
		unarm(&h.cfg)
		if sourceFinalArmed(h.cfg) {
			t.Fatalf("%s: armed", name)
		}
		h.sourcePass(t)
		h.untouched(t, name)
		h.cfg = saved
	}
	var prod moveCfg
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
	armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
	if prod.DeleteHeld == nil || prod.Guard.DataDir == "" || prod.Local.DataDir == "" || prod.Versioned == nil {
		t.Fatalf("production wiring: %+v", prod)
	}
	prod.Route = func(string) (*registry.Route, error) { return nil, nil }
	if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || sourceFinalArmed(prod) {
		t.Fatal("production armed the finalization")
	}
}

// The rename helper on its own: absent candidates, a refused path, and a
// destination that holds something else.
func TestParkLocalArtifactsRefusals(t *testing.T) {
	dd := t.TempDir()
	j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
	mark := func(step, value string) error { j.Mark(step, value); return nil }
	p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", NewID: func() string { return "r1" }}
	for _, bad := range []string{"../x", "/abs", ".parked/x", "hostguard", "a/../b", ""} {
		p.Paths = []string{bad}
		if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
			t.Errorf("%q was accepted", bad)
		}
	}
	if len(j.Steps) != 0 {
		t.Fatalf("a refused park journalled %v", j.Steps)
	}
	if err := os.WriteFile(filepath.Join(dd, "a"), []byte("a"), 0o600); err != nil {
		t.Fatal(err)
	}
	p.Paths = []string{"a", "absent"}
	p.Fault = func(point string) error {
		if point == "after-rename:a" {
			return errCrash
		}
		return nil
	}
	if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
		t.Fatalf("crash: %v", err)
	}
	// Something else at the destination stops the park; nothing more moves.
	dst := filepath.Join(parkedArtifactsDir(dd, "inc", p1Host, "r1"), "a")
	if err := os.Rename(dst, filepath.Join(dd, "a")); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(dst, []byte("other"), 0o600); err != nil {
		t.Fatal(err)
	}
	p.Fault = nil
	if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
		t.Fatal("a foreign destination was accepted")
	}
	if b, _ := os.ReadFile(filepath.Join(dd, "a")); string(b) != "a" {
		t.Fatalf("the live artifact moved: %q", b)
	}
	os.Remove(dst)
	if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
		t.Fatalf("resume: %q %v", r, err)
	}
	if _, err := os.Lstat(filepath.Join(dd, "a")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a is still live: %v", err)
	}
	// Another plan under the same step is refused.
	p.Paths = []string{"b"}
	j.Unmark("s")
	if _, err := parkLocalArtifacts(context.Background(), j, mark, p); err == nil {
		t.Fatal("a plan for other paths was accepted")
	}
}

// testSourcePaths is what the tests expect the source to keep at its flat
// live paths and park: the database with its marker, journal, quarantine
// and litestream shadow, the files directory, <host>.env, and the two
// rendered env files (fix round 1, ruling 3).
func testSourcePaths() []string {
	db := p1Host + ".db"
	return []string{db, db + ".gen", db + "-wal", db + "-shm", db + ".corrupt", ".activator-" + db + "-litestream", p1Host, p1Host + ".env",
		p1Host + ".config.env", p1Host + "-sidecar.env"}
}

// sourceTestView and sourceTestAuthorize are the park view and authorizer
// the tests inject for the source's location park, in place of Plan 5's
// owners (fix round 1, ruling 4: production never substitutes stand-ins).
// The target's region is authoritative after the commit, so every version
// under the location is selected; a chore is authorized only when it is
// this move's, for ie-box or ie, under the move's committed landing with
// ie-box its source.
func sourceTestView(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
	return sourceTestViewFor("ie", "uk")(ctx, obs)
}

// sourceTestViewFor is sourceTestView for a source seen from region with
// the target's region authoritative (the harness's sides: intoHome's is ca
// with @home authoritative).
func sourceTestViewFor(region, authoritative string) func(context.Context, hostclaim.Observed) (park.View, error) {
	return func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
		if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
			return park.View{}, park.ErrUnsafe
		}
		location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
		return park.View{ClaimETag: obs.ETag, Region: region, AuthoritativeRegion: authoritative,
			GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
	}
}

// sourceTestLocalAuthorize is the claim's local authorizer (M9) for the
// source's park-local: sourceTestAuthorize over the claim park.Local read.
func sourceTestLocalAuthorize(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
	return sourceTestLocalAuthorizeFor("ie-box", "ie")(ctx, c, o)
}

// sourceTestLocalAuthorizeFor is sourceTestLocalAuthorize for the source
// box and its engine region.
func sourceTestLocalAuthorizeFor(box, region string) func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error {
	authorize := sourceTestAuthorizeFor(box, region)
	return func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
		return authorize(ctx, hostclaim.Observed{Host: p1Host, Claim: c}, o)
	}
}

func sourceTestAuthorize(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
	return sourceTestAuthorizeFor("ie-box", "ie")(ctx, obs, o)
}

// sourceTestAuthorizeFor is sourceTestAuthorize for the source box and its
// engine region.
func sourceTestAuthorizeFor(box, region string) func(context.Context, hostclaim.Observed, hostclaim.Obligation) error {
	return func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
		c := obs.Claim
		if obs.Absent || c == nil || c.Op.ID != p1ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted ||
			len(c.Op.Participants) != 2 || c.Op.Participants[0] != box || o.Op != p1ID {
			return park.ErrUnsafe
		}
		if (o.Kind == "park-local" && o.Box == box) || (o.Kind == "park-replica" && o.Region == region) {
			return nil
		}
		return park.ErrUnsafe
	}
}

// Fix round 1, ruling 1: a crash after source_parked_at is stamped and
// before the journal is parked, then the close (which clears the move, so
// no record routes here any more), then a fresh source pass: the local
// sweep parks the journal, because the close proves the stamp landed.
func TestSourceParkJournalSurvivesTheClose(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	h.finalAt = "after-source-parked-record"
	h.crashSource(t)
	if j := h.sourceJournal(t); j.Done(finalStepParked) && j.Steps[finalStepParked] == "done" {
		t.Fatalf("fixture: the stamp is journalled done: %v", j.Steps)
	}
	ctx := context.Background()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	for _, mut := range []func(*provision.Record) error{provision.MoveOpened(g, h.clock.Now()), provision.MoveClose(g, h.clock.Now())} {
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, mut); err != nil {
			t.Fatal(err)
		}
	}
	if rec := h.read(t, p1Host); rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID {
		t.Fatalf("fixture: not closed: %+v", rec)
	}
	if movesParticipantRecord(h.cfg, h.read(t, p1Host)) != roleNone {
		t.Fatal("fixture: a closed record still routes here")
	}
	h.freshWorker()
	h.sourcePass(t)
	if got := h.live(t); len(got) != 0 {
		t.Fatalf("still live after the close: %v", got)
	}
	j := h.parkedJournal(t)
	if j.Steps[finalStepParked] != "done" {
		t.Fatalf("the parked journal's stamp: %v", j.Steps)
	}
}

// The sweep never touches a journal that has not reached the stamp, nor one
// whose stamp is pending while the record still carries the op unstamped: a
// journal of another op (a landing's, say) under a record with no move, and
// this op's journal stopped right after it journalled the stamp pending,
// before the stamp's CAS.
func TestSourceParkSweepLeavesUnstampedJournals(t *testing.T) {
	t.Parallel()
	h := newFinalHarness(t)
	h.commit(t)
	// Another op's journal that would park cleanly if the sweep took it: a
	// complete local plan and retirement, everything but the stamp step.
	plan, err := json.Marshal(localParkPlan{Retirement: "r0", Incarnation: h.inc})
	if err != nil {
		t.Fatal(err)
	}
	otherSteps := map[string]string{finalStepLocal + "-plan": string(plan), finalStepLocal: "r0", finalStepRowDeleted: "done"}
	other, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2)
	if err != nil {
		t.Fatal(err)
	}
	for k, v := range otherSteps {
		other.Mark(k, v)
	}
	if err := other.Commit(h.cfg.DataDir); err != nil {
		t.Fatal(err)
	}
	h.finalAt = "after-source-parked-pending"
	h.crashSource(t)
	if j := h.sourceJournal(t); j.Steps[finalStepParked] != "pending" || h.read(t, p1Host).Move.SourceParkedAt != nil {
		t.Fatalf("fixture: %v", j.Steps)
	}
	h.cfg.ParkLeases = nil // unarmed: only the sweep runs
	h.freshWorker()
	h.sourcePass(t)
	for _, op := range []string{p1ID, p1ID2} {
		if !moveJournalPresent(h.cfg.DataDir, p1Host, op) {
			t.Fatalf("the sweep took %s's journal before its stamp", op)
		}
	}
	if j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID2); err != nil || len(j.Steps) != len(otherSteps) {
		t.Fatalf("the sweep wrote into another op's journal: %v %v", j.Steps, err)
	}
}

// Fix round 1, ruling 2: a replay that finds an artifact already at its
// destination (a stop after the rename, before either sync) syncs the
// source's parent and every destination parent before it goes on; and the
// rename offers an interruption point between the rename and the syncs.
func TestParkLocalArtifactsSyncsOnReplay(t *testing.T) {
	dd := t.TempDir()
	if err := os.MkdirAll(filepath.Join(dd, "d", "e"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(dd, "d", "e", "a"), []byte("a"), 0o600); err != nil {
		t.Fatal(err)
	}
	j := &moves.Journal{Host: p1Host, Op: p1ID, Steps: map[string]string{}}
	mark := func(step, value string) error { j.Mark(step, value); return nil }
	var points []string
	p := localArtifactPark{DataDir: dd, Incarnation: "inc", Host: p1Host, Op: p1ID, Step: "s", Paths: []string{"d/e/a"}, NewID: func() string { return "r1" },
		Fault: func(point string) error {
			points = append(points, point)
			if point == "after-rename-unsynced:d/e/a" {
				return errCrash
			}
			return nil
		}}
	if _, err := parkLocalArtifacts(context.Background(), j, mark, p); !errors.Is(err, errCrash) {
		t.Fatalf("no interruption between the rename and its syncs: %v (points %v)", err, points)
	}
	var synced []string
	saved := retireSync
	retireSync = func(path string) error { synced = append(synced, path); return saved(path) }
	t.Cleanup(func() { retireSync = saved })
	p.Fault = nil
	if r, err := parkLocalArtifacts(context.Background(), j, mark, p); err != nil || r != "r1" {
		t.Fatalf("replay: %q %v", r, err)
	}
	art := parkedArtifactsDir(dd, "inc", p1Host, "r1")
	for _, want := range []string{filepath.Join(dd, "d", "e"), filepath.Join(art, "d", "e"), filepath.Join(art, "d"), art, filepath.Dir(art), filepath.Join(dd, ".parked")} {
		found := false
		for _, s := range synced {
			found = found || s == want
		}
		if !found {
			t.Errorf("the replay did not sync %s (synced %v)", want, synced)
		}
	}
}