//go:build linux

package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfence"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// Plan 5 Task 9b's integrated acceptance tests (ruling Q12): claim Plan 12
// proved its half with claim fixtures (TestPlan12RecoveryAdmissionAfter
// TargetSession, TestPlan12RecoveryRestoreSourceDatabase); these run the
// same paths with the moves owners the registered wake source opens: the
// claim's wake runtime, the cross-box handover, the claim's writer with
// the moves owners' evidence and loss owner, and, where the database is
// restored, ColdRestore and the source installation's retirement.

// mwAccess is the claim's move read access over a poFixture: its route and
// launch as they are when asked, and the held works the fixture selects.
type mwAccess struct{ f *poFixture }

func (a mwAccess) HeldWork(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error) {
	return a.f.owners.Works(ctx, host, action, captured)
}
func (a mwAccess) Route(_ context.Context, host string) (*registry.Route, error) {
	if host != a.f.route.Host {
		return nil, hostbox.ErrChanged
	}
	rt := a.f.route
	return &rt, nil
}
func (a mwAccess) Launch(_ context.Context, host string) (ClaimServingLaunch, moves.EnvReport, error) {
	if host != a.f.route.Host {
		return ClaimServingLaunch{}, moves.EnvReport{}, hostbox.ErrChanged
	}
	// The env report of the same launch env, as claimMoveLaunchSource makes
	// it: its digest is the launch's move-portable pin.
	key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
	if err != nil {
		return ClaimServingLaunch{}, moves.EnvReport{}, err
	}
	return a.f.launch, moves.ReportLaunch(key, launchVars(a.f.launch.Env), nil), nil
}

// mwWorld is one box's registered moves wake source over a poFixture's
// world, and the claim's wake runtime that box builds around it.
type mwWorld struct {
	t           *testing.T
	f           *poFixture
	box, region string
	local       claimStopLocal
	owners      ClaimMovesOwners
	m           *movesProbeOwners
	files       ClaimFilesManifest
	guard       hostguard.Guard
	journal     *hostbox.LaunchJournal
	handover    claimCrossBoxHandover
	wake        *ClaimWakeAssembly
	runtime     *activator.ClaimRuntime
}

// mwRegister registers the moves wake source as the agent does (the parks,
// then the wake source, both switches on), opens it as the claim's
// OpenClaimOwners does for box in region (the source's Open, then the
// claim's C8 provider through openClaimFilesManifest), and repins the
// fixture's recorded environment reports under the registered source's key
// (CARLOS_SECRETS_KEY's). Only the records and the marker store are the
// fixture's: the fixture keeps its record in memory and its bucket in a
// fake.
func mwRegister(t *testing.T, f *poFixture, box, region string, files ClaimFilesManifest) *mwWorld {
	t.Helper()
	ctx := t.Context()
	withMovesWakeSource(t, false)
	withClaimFilesSource(t, nil)
	parks, err := newMovesParkSources(f.owners.Records, f.bucket, "data", fixedRegion(region), movesRetirementEvidence{Records: f.owners.Records, Now: time.Now})
	if err != nil {
		t.Fatal(err)
	}
	if err := registerMovesParks(parks); err != nil {
		t.Fatal(err)
	}
	env := mwEnv(true, true)
	if err := registerProductionMovesWakeSource(env, parks, f.dataDir); err != nil {
		t.Fatal(err)
	}
	if err := movesWakeSourceRegistered(); err != nil {
		t.Fatal(err)
	}
	// The app's manifest, which the registered owners read for the
	// readiness path, and the files directory where production keeps it
	// (movesFilesDirFor: <data>/<host>); a rename keeps its identity.
	if err := provision.PutManifest(ctx, f.claims, "bab", "jam", provision.Manifest{Moves: &provision.MovesManifest{ReadyPath: "/healthz"}}); err != nil {
		t.Fatal(err)
	}
	if dir, _ := movesFilesDirFor(f.dataDir)(poHost); f.files != dir {
		if err := os.Rename(f.files, dir); err != nil {
			t.Fatal(err)
		}
		f.files = dir
	}
	src := claimMoveSources.Source.(*movesWakeSourceRegistration).ClaimProductionMoveSources.(*movesWakeSources)
	records := f.owners.Records
	src.records = func(release.Store) MovesRecordReader { return records }
	src.markers = func(context.Context, string, string) (ClaimMarkerStore, error) { return plan9D5Markers{f.bucket}, nil }
	src.now, src.newID = f.owners.Now, f.owners.NewID
	w := &mwWorld{t: t, f: f, box: box, region: region}
	// The box's local guard and parked area under the fixture's data
	// directory.
	for _, dir := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(f.dataDir, dir), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	w.guard = hostguard.Guard{DataDir: f.dataDir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: f.claims}}
	w.local = claimStopLocal{Actor: hostbox.Actor{Box: box, Enrollment: "enrollment-1", Revision: 7, Binding: "7", Region: region}, BootID: "boot-1", ControllerID: "controller-1", PID: 4101, ProcessStart: "start-1"}
	actor := hostbox.Actor{Box: box, Region: region, Bucket: "data", Enrollment: "enrollment-1", Cloud: hostbox.CloudIdentity{Region: "eu-west-1"}}
	out := ClaimOwnerBindings{Data: f.bucket}
	var opened ClaimFilesManifest
	// The claim hands Open its bucket through the request store
	// (openProductionClaimMoveWake), with the box's identity check.
	data := hostbox.RequestStore{Store: f.bucket, Check: func(context.Context) error { return nil }}
	out.Moves, opened, err = claimMoveSources.Source.Open(ctx, f.claims, data, actor, mwAccess{f})
	if err != nil || opened != nil {
		t.Fatalf("the registered source's Open: %v (files %v)", err, opened)
	}
	if _, err := openClaimFilesManifest(ctx, env, &fakeFilesSource{files: files}, f.claims, f.bucket, actor, &out, func(context.Context) error { return nil }); err != nil {
		t.Fatal(err)
	}
	w.owners, w.files = out.Moves, out.Serving.Files
	w.m = out.Moves.Inputs.(*movesProbeOwners)
	if w.m.Files != files || w.files != files {
		t.Fatalf("the recovery owners' C8 provider %v is not the claim's %v", w.m.Files, w.files)
	}
	key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
	if err != nil {
		t.Fatal(err)
	}
	if mv := f.rec.Move; mv != nil {
		if mv.Inputs != nil {
			mv.Inputs.EnvReport = envDigest(key, f.launch.Env)
		}
		if mv.RecoveryInputs != nil {
			mv.RecoveryInputs.EnvReport = localEnvDigest(key, f.launch.Env)
		}
		if mv.Rollback != nil && mv.Rollback.RecoveryInputs != nil {
			mv.Rollback.RecoveryInputs.EnvReport = localEnvDigest(key, f.launch.Env)
		}
	}
	return w
}

// build is the claim's wake runtime on this box (claimconstruction.go's
// shape): the moves owners behind ClaimOwnerDispatch, the claim's writer
// with their evidence and the moves loss owner (pub is the loss records'
// root key), the cross-box handover over real registry rows and stop
// inventory, and the source installation's retirement through park.Local.
func (w *mwWorld) build(pub ed25519.PublicKey, inventory *hostbox.StopInventoryDB, reg *registry.Registry) {
	t, f := w.t, w.f
	journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { journal.Close() })
	w.journal = journal
	dispatch := ClaimOwnerDispatch{Moves: &w.owners.ClaimKindOwners}
	local, now := w.local, f.owners.Now
	producers := &ClaimProducerSet{}
	retirement := claimArtifactRetirement{Local: park.Local{Guard: w.guard, Claims: f.claims, DataDir: f.dataDir, Box: w.box}, Continue: hostclaim.SessionContinues}
	wk := &ClaimWakeAssembly{LeaseRegion: "eu-west-1", Inputs: dispatch, Manager: new(cgroup.Manager), Operations: w.owners.Operations, Continue: hostclaim.SessionContinues}
	wk.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
	wk.AfterAdmission = func(context.Context, hostclaim.Observed, hostclaim.Observed) error { return nil }
	// Every predecessor here is a remote one (the cross-box handover) or
	// none; root's local stop is never asked.
	wk.Stop = func(context.Context, hostclaim.Observed) error { return hostbox.ErrChanged }
	wk.PrepareState = func(ctx context.Context, _ hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) error {
		return work.With(ctx, in.Identity, func() error { return os.MkdirAll(filepath.Dir(in.DBPath), 0o700) })
	}
	wk.RetireArtifacts, wk.RecoverArtifacts = retirement.Retire, retirement.Recover
	wk.Retain = func(hostclaim.Observed, activator.Proc, activator.Rep, error) {}
	wk.ObserveLoss = func(hostclaim.Observed, error) {}
	wk.Producers = func(_ context.Context, o hostclaim.Observed) (*ClaimProducers, error) {
		return producers.Ensure(recoveryTuple(o, local), local.ProcessStart, journal.RecordProducer)
	}
	wk.Launch = func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error) {
		return nil, hostwrite.ErrUnavailable
	}
	wk.Prelaunch = func(ctx context.Context, o hostclaim.Observed, work hostwrite.Work, in ClaimLaunchInputs) (*activator.RoleLaunch, error) {
		target := recoveryTuple(o, local)
		err := work.With(ctx, in.Identity, func() error {
			return journal.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: local.PID, ProcessStart: local.ProcessStart, Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: target.Session}}, NoState: true})
		})
		if err != nil {
			return nil, err
		}
		return &activator.RoleLaunch{Target: target, Journal: journal}, nil
	}
	wk.Retries = activator.ServingRetries{Claims: f.claims, Box: w.box, Enrollment: local.Actor.Enrollment, BootID: local.BootID, ControllerID: local.ControllerID, Now: now, NewID: f.owners.NewID,
		Inputs:      func(context.Context, registry.Route) (string, error) { return "inputs", nil },
		StopAttempt: func(context.Context, hostclaim.Observed, hostclaim.WakeRetryAttempt) error { return hostbox.ErrChanged }}
	w.handover = claimCrossBoxHandover{Guard: w.guard, Evidence: f.claims, Box: w.box, PublicKey: pub, FenceAuthority: plan12FenceAuthority{}, Identity: handoverRecoveryIdentity(t, local.Actor), Inventory: inventory, Registry: reg}
	wireClaimAdmission(wk, w.box, wk.AfterAdmission, w.handover, nil, nil)
	source := hostbox.WorkSource{Guard: w.guard, Box: w.box, Region: w.region, Now: now,
		Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil },
		Local:  func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
	writer := hostbox.Writer{Data: f.bucket, Bucket: "data", Claims: f.claims, Box: w.box, Region: w.region, AdmissionEvidence: dispatch.AdmissionEvidence,
		Loss: &hostbox.LossAdmission{PublicKey: pub, Owner: w.owners.LossRecovery, Sources: w.owners.LossSources}}
	runtime, err := wk.Build(t.Context(), source, writer)
	if err != nil {
		t.Fatal(err)
	}
	w.wake, w.runtime = wk, runtime
}

// admit stands in for Prerequisites' S3 lease, then admits through the
// runtime: the moves owners' Select, the handover's recovery intent, the
// claim's writer and the moves owners' evidence, the handover after the
// CAS, the guard's activation and the pre-launch record.
func (w *mwWorld) admit(o hostclaim.Observed, p activator.StartPolicy) (hostclaim.Observed, error) {
	w.t.Helper()
	live, cancel := context.WithCancel(context.Background())
	w.t.Cleanup(cancel)
	done := make(chan struct{})
	go func() { <-live.Done(); close(done) }()
	w.wake.mu.Lock()
	w.wake.pending[o.Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
	w.wake.mu.Unlock()
	return w.runtime.Admit(w.t.Context(), w.f.route, o, p)
}

// mwSnapshot is a real Litestream snapshot (one level-9 LTX file, TXIDs 1
// to maxTXID) of a SQLite database holding value.
func mwSnapshot(t *testing.T, maxTXID uint64, value string) []byte {
	t.Helper()
	path := filepath.Join(t.TempDir(), "snapshot.db")
	db, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec("CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES (?)", value); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(); err != nil {
		t.Fatal(err)
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	size := int(binary.BigEndian.Uint16(raw[16:18]))
	if size == 1 {
		size = 65536
	}
	spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(size), Commit: uint32(len(raw) / size), MinTXID: 1, MaxTXID: ltx.TXID(maxTXID)}}
	for off := 0; off < len(raw); off += size {
		pg := uint32(off/size + 1)
		if pg != ltx.LockPgno(uint32(size)) {
			spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pg}, Data: raw[off : off+size]})
		}
	}
	var out bytes.Buffer
	if _, err := spec.WriteTo(&out); err != nil {
		t.Fatal(err)
	}
	return out.Bytes()
}

// mwReadDB reads the fixture value of the database at path.
func mwReadDB(t *testing.T, path string) string {
	t.Helper()
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var value string
	if err := db.QueryRow("SELECT value FROM fixture").Scan(&value); err != nil {
		t.Fatalf("read %s: %v", path, err)
	}
	return value
}

// mwSourceInstallation is the source's installation at S on ie-box as the
// cut and the return leave it, for the handover and the retirement: the
// source's stopped report (the cut's evidence), the guard's local record at
// S, the database and its marker on disk tagged S in the claim's inventory,
// and the registry row (with a timer) and complete stop inventory at S.
type mwSourceInstallation struct {
	inventory *hostbox.StopInventoryDB
	registry  *registry.Registry
	dbRel     string
}

func (w *mwWorld) sourceInstallation() mwSourceInstallation { return w.sourceInstallationRow(rcS) }

// sourceInstallationRow is sourceInstallation with the registry's route row
// owned by the installation at rowStamp.
func (w *mwWorld) sourceInstallationRow(rowStamp uint64) mwSourceInstallation {
	t, f := w.t, w.f
	ctx := t.Context()
	o := f.read()
	h := o.Claim.Holder
	report := moves.StoppedReport{V: 1, Role: "source", Box: w.box, Host: poHost, OpID: poOp, Incarnation: h.Incarnation, ClaimGen: o.Claim.Gen, CheckedAt: time.Now().UTC(),
		Steps: map[string]moves.Check{moves.StepTree: {OK: true}, moves.StepStray: {OK: true}, moves.StepFDs: {OK: true}, moves.StepLease: {OK: true}, moves.StepReplica: {OK: true}}}
	raw, _ := json.Marshal(report)
	if _, err := f.claims.Put(ctx, moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "source"), raw, ""); err != nil {
		t.Fatal(err)
	}
	if err := w.guard.Activate(ctx, poHost, w.box, nil); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(f.route.DBPath, []byte("source database"), 0o600); err != nil {
		t.Fatal(err)
	}
	dbRel, err := filepath.Rel(f.dataDir, f.route.DBPath)
	if err != nil {
		t.Fatal(err)
	}
	local := park.Local{Guard: w.guard, Claims: f.claims, DataDir: f.dataDir, Box: w.box}
	if err := local.RecordInventory(ctx, poHost, hostguard.Captured{Incarnation: h.Incarnation, Stamp: rcS}, "flat", []string{dbRel, dbRel + ".gen"}, func(*hostclaim.Claim) error { return nil }); err != nil {
		t.Fatal(err)
	}
	inv, err := openLegacyInventory(t, filepath.Join(t.TempDir(), "inventory.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { inv.Close() })
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	row := registry.Route{Host: poHost, Kind: "instance", DBPath: f.route.DBPath, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: h.Incarnation, ClaimStamp: rowStamp, ClaimOwner: "old-owner", ClaimRowID: "old-row"}}
	if err := reg.Upsert(row); err != nil {
		t.Fatal(err)
	}
	// A timer takes its route's ownership, under a row id of its own.
	timer, err := reg.UpsertTimerRegistration(registry.Timer{Host: poHost, Name: "nightly", Kind: "cron", Spec: "0 3 * * *"}, "")
	if err != nil {
		t.Fatal(err)
	}
	if err := inv.WriteStopInventory(ctx, hostbox.StopInventory{Complete: true, Host: poHost, Incarnation: h.Incarnation, InstallationStamp: rcS, Box: w.box, Enrollment: w.local.Actor.Enrollment, DirectoryRevision: "7",
		Entries: []hostbox.StopInventoryEntry{
			{Kind: "route", ArtifactID: "route", ManagedRoot: "registry", RelativePath: poHost, RowIdentity: "old-row", Ownership: "old-owner", Materialization: "materialized"},
			{Kind: "timer", ArtifactID: "timer", ManagedRoot: "registry", RelativePath: poHost + "/nightly", RowIdentity: timer.ClaimRowID, Ownership: "old-owner", Materialization: "materialized"},
		}}); err != nil {
		t.Fatal(err)
	}
	return mwSourceInstallation{inventory: inv, registry: reg, dbRel: dbRel}
}

// targetStopped journals the target's stop evidence for its last session
// (the claim's target journal, J3): the session, the generation, and the
// target's stopped report.
func (w *mwWorld) targetStopped() {
	t, f := w.t, w.f
	o := f.read()
	h := o.Claim.Holder
	last := h.Admissions[len(h.Admissions)-1]
	key := moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "target")
	report := moves.StoppedReport{V: 1, Role: "target", Box: last.Box, Host: poHost, OpID: poOp, Incarnation: h.Incarnation, ClaimGen: o.Claim.Gen, CheckedAt: time.Now().UTC(),
		Steps: map[string]moves.Check{moves.StepTree: {OK: true}, moves.StepStray: {OK: true}, moves.StepFDs: {OK: true}, moves.StepLease: {OK: true}, moves.StepReplica: {OK: true}}}
	raw, _ := json.Marshal(report)
	if _, err := f.claims.Put(t.Context(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
	c := *o.Claim
	c.Op.Journal[hostclaim.JournalEvidenceTarget] = key
	c.Op.Journal[hostclaim.JournalTargetSession] = strconv.FormatUint(last.Session, 10)
	c.Op.Journal[hostclaim.JournalTargetGeneration] = last.Generation
	f.putClaim(&c)
}

// srcSnapshot replaces src_gen's root with a real snapshot holding value at
// the source's maximum (30), the clean return's restore point.
func (w *mwWorld) srcSnapshot(value string) {
	w.f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), mwSnapshot(w.t, 30, value))
}

// restore prepares the admitted start and runs its restore.
func (w *mwWorld) restore(admitted hostclaim.Observed, p activator.StartPolicy) activator.ClaimStart {
	w.t.Helper()
	s, err := w.runtime.Prepare(w.t.Context(), w.f.route, admitted, p)
	if err != nil {
		w.t.Fatal(err)
	}
	if err := s.Restore(w.t.Context()); err != nil {
		w.t.Fatal(err)
	}
	return s
}

// TestRecoveryAdmissionAfterTargetSession (Q12; claim Plan 12's
// TestPlan12RecoveryAdmissionAfterTargetSession with the registered moves
// owners): the source's first recovery admission after a target session,
// on each of the three authorities a remote predecessor can have, the
// target's journalled stop, a declared loss (J4) and a consumed
// termination fence, goes through the handover, the claim's writer under
// the moves owners' evidence and loss owner, and the restore: the source
// executes a new session over the restored source baseline.
func TestRecoveryAdmissionAfterTargetSession(t *testing.T) {
	for _, mode := range []string{"journal", "loss", "fence"} {
		t.Run(mode, func(t *testing.T) {
			f := newRCFixture(t, rcClean)
			var pub ed25519.PublicKey
			switch mode {
			case "loss":
				pub = f.declare(false, []hostclaim.WriteSource{rcTargetSource(poOp, 20), rcCutSource()}, nil)
			case "fence":
				f.fence()
			}
			w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
			w.srcSnapshot("the source baseline")
			src := w.sourceInstallation()
			w.build(pub, src.inventory, src.registry)
			if mode == "journal" {
				w.targetStopped()
			}
			o := f.read()
			before := o.Claim.Holder
			if !claimRemotePredecessor(o, "ie-box") || before.WriterBox != "uk-box" || before.Session != 20 {
				t.Fatalf("the predecessor is not the target's session: %+v", before)
			}
			if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
				t.Fatalf("the predecessor's authority: %v", err)
			}
			admitted, err := w.admit(o, f.policy())
			if err != nil {
				t.Fatal(err)
			}
			h := admitted.Claim.Holder
			a := h.Admissions[len(h.Admissions)-1]
			if h.WriterBox != "ie-box" || h.Session <= 20 || h.Stamp != h.Session || h.Generation != "fresh1" || a.Box != "ie-box" || a.Region != "ie" || a.Mode != "cold" || a.Predecessor != poOp || a.PredecessorRegion != "uk" {
				t.Fatalf("the admission: holder %+v, last %+v", h, a)
			}
			switch mode {
			case "journal":
				if a.Coverage != recoveryCoverageClean || a.LossID != "" || a.WritesLost != nil {
					t.Fatalf("a stopped target's clean return: %+v", a)
				}
			case "loss":
				if a.Coverage != "" || a.LossID != rcLossID || a.WritesLost == nil || a.WritesLost.Box != "uk-box" || a.WritesLost.Session != 20 || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) {
					t.Fatalf("a declared loss's admission: %+v", a)
				}
			case "fence":
				if a.Coverage != "" || a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) {
					t.Fatalf("a fenced target's admission: %+v", a)
				}
			}
			w.restore(admitted, f.policy())
			if got, err := os.ReadFile(f.route.DBPath + ".gen"); err != nil || string(got) != h.Generation+"\n" {
				t.Fatalf("the restored marker: %q %v", got, err)
			}
			if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
				t.Fatalf("restored %q", got)
			}
		})
	}
}

// TestCleanReturnAdmissionRetiresTheSTaggedDatabase (Q12): the clean
// return's first admission restores src_gen over the source's own
// installation, whose database and marker the claim's inventory tags with
// the source stamp S. The restore retires them, on the cut's stop
// evidence, into the superseded installation's parked artifacts, and the
// claim's inventory then tags the paths with the admitted stamp; a restart
// restores the same pinned baseline and retires nothing further.
func TestCleanReturnAdmissionRetiresTheSTaggedDatabase(t *testing.T) {
	f := newRCFixture(t, rcClean)
	w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
	w.srcSnapshot("the source baseline")
	src := w.sourceInstallation()
	w.build(nil, src.inventory, src.registry)
	w.targetStopped()
	o := f.read()
	if tag := o.Claim.Inventory["ie-box"].Paths[src.dbRel]; tag != rcS {
		t.Fatalf("the source database is tagged %d, not S", tag)
	}
	if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
		t.Fatal(err)
	}
	admitted, err := w.admit(o, f.policy())
	if err != nil {
		t.Fatal(err)
	}
	h := admitted.Claim.Holder
	w.restore(admitted, f.policy())
	retired := filepath.Join(f.dataDir, ".parked", h.Incarnation, poHost, fmt.Sprintf("superseded-%d-%d", h.Session, rcS), "artifacts", src.dbRel)
	if got, err := os.ReadFile(retired); err != nil || string(got) != "source database" {
		t.Fatalf("the S-tagged database was not retired: %q %v", got, err)
	}
	if got, err := os.ReadFile(retired + ".gen"); err != nil || string(got) != "src\n" {
		t.Fatalf("the S-tagged marker was not retired: %q %v", got, err)
	}
	if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
		t.Fatalf("restored %q", got)
	}
	after := f.read()
	for _, p := range []string{src.dbRel, src.dbRel + ".gen"} {
		if tag := after.Claim.Inventory["ie-box"].Paths[p]; tag != h.Stamp {
			t.Fatalf("%s is tagged %d after the restore, want the admitted stamp %d", p, tag, h.Stamp)
		}
	}
	// The recorded source baseline pins the restore: a newer version of
	// src_gen's root is refused, and nothing more is retired.
	w.f.bucket.Put("data", "instances/"+poHost+"/g-src/0009/"+ltx.FormatFilename(1, 30), mwSnapshot(t, 30, "a newer write"))
	if _, err := w.m.Select(t.Context(), f.route, after, f.policy()); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("a restart over another version of src_gen: %v", err)
	}
}

// TestDirtyReturnThroughAfterRecoveryOnRealRows (Q12, the rev 3 recheck
// §2.3): a dirty return's first admission, into the return generation,
// goes through the handover's afterRecovery over the box's real registry
// rows (the route and a timer) and its stop inventory at S: the rows and
// the inventory move to the admitted installation's ownership, the
// source's inventory stays for retirement and replay, and a row another
// installation owns refuses the admission's handover.
func TestDirtyReturnThroughAfterRecoveryOnRealRows(t *testing.T) {
	t.Run("the rows move to the admitted installation", func(t *testing.T) {
		f := newRCFixture(t, rcDirty)
		w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
		src := w.sourceInstallation()
		w.build(nil, src.inventory, src.registry)
		w.targetStopped()
		o := f.read()
		if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
			t.Fatal(err)
		}
		admitted, err := w.admit(o, f.policy())
		if err != nil {
			t.Fatal(err)
		}
		h := admitted.Claim.Holder
		a := h.Admissions[len(h.Admissions)-1]
		if h.Generation != rpRB || h.Stamp == rcS || h.Stamp != h.Session || a.Coverage != "sealed" || a.Predecessor != poOp {
			t.Fatalf("the dirty return's admission: holder %+v, last %+v", h, a)
		}
		sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", poHost, h.Incarnation, h.Stamp, "ie-box")))
		owner := hex.EncodeToString(sum[:])
		row, err := src.registry.Get(poHost)
		if err != nil || row.ClaimStamp != h.Stamp || row.ClaimOwner != owner || row.ClaimIncarnation != h.Incarnation || row.ClaimRowID == "old-row" {
			t.Fatalf("the route row: %+v %v", row, err)
		}
		timers, err := src.registry.TimersFor(poHost)
		if err != nil || len(timers) != 1 || timers[0].ClaimStamp != h.Stamp || timers[0].ClaimOwner != owner || timers[0].ClaimRowID == "" || timers[0].ClaimRowID == row.ClaimRowID {
			t.Fatalf("the timer row: %+v %v", timers, err)
		}
		inv, err := src.inventory.ReadInstallation(t.Context(), poHost, h.Incarnation, h.Stamp)
		if err != nil || !inv.Complete || inv.Box != "ie-box" || len(inv.Entries) != 2 {
			t.Fatalf("the admitted installation's inventory: %+v %v", inv, err)
		}
		for _, e := range inv.Entries {
			if e.Ownership != owner || (e.Kind == "route" && e.RowIdentity != row.ClaimRowID) || (e.Kind == "timer" && e.RowIdentity != timers[0].ClaimRowID) {
				t.Fatalf("an inventory entry: %+v", e)
			}
		}
		if old, err := src.inventory.ReadInstallation(t.Context(), poHost, h.Incarnation, rcS); err != nil || !old.Complete {
			t.Fatalf("the source's inventory at S is gone: %v", err)
		}
	})
	t.Run("a row another installation owns refuses the handover", func(t *testing.T) {
		f := newRCFixture(t, rcDirty)
		w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
		src := w.sourceInstallationRow(99)
		w.build(nil, src.inventory, src.registry)
		w.targetStopped()
		o := f.read()
		if err := w.wake.stopPredecessor(t.Context(), o); err != nil {
			t.Fatal(err)
		}
		if _, err := w.admit(o, f.policy()); !errors.Is(err, hostbox.ErrTypedInventory) {
			t.Fatalf("a foreign row: %v", err)
		}
		if got, err := src.registry.Get(poHost); err != nil || got.ClaimStamp != 99 || got.ClaimRowID != "old-row" {
			t.Fatalf("the foreign row moved: %+v %v", got, err)
		}
	})
}

// TestCommittedColdWakeThroughTheRegisteredSource (Q12): after the
// landing's probe passed and the move committed, a wake on the target is a
// cold admission into a fresh generation from the certified remote, through
// the registered source's owners and the claim's runtime: the moves
// owners' evidence admits it, and the start runs the committed wake's
// ordinary launch, no probe variable, bound to the moves attempt journal.
func TestCommittedColdWakeThroughTheRegisteredSource(t *testing.T) {
	f, o := committedAfterProbe(t)
	w := mwRegister(t, f, "uk-box", "uk", &rcFiles{f: f})
	w.build(nil, nil, nil)
	if err := w.guard.Activate(t.Context(), poHost, "uk-box", nil); err != nil {
		t.Fatal(err)
	}
	h := o.Claim.Holder
	if h.WriterBox != "uk-box" || h.Generation != poOp {
		t.Fatalf("not the committed landing: %+v", h)
	}
	admitted, err := w.admit(o, poWake)
	if err != nil {
		t.Fatal(err)
	}
	ah := admitted.Claim.Holder
	a := ah.Admissions[len(ah.Admissions)-1]
	if ah.Generation == poOp || ah.Generation != a.Generation || a.Mode != "cold" || a.Coverage != "" || a.Predecessor != poOp || a.Region != "uk" || a.PredecessorRegion != "uk" || ah.WriterBox != "uk-box" || ah.Session <= h.Session {
		t.Fatalf("the committed admission: holder %+v, last %+v", ah, a)
	}
	s, err := w.runtime.Prepare(t.Context(), f.route, admitted, poWake)
	if err != nil {
		t.Fatal(err)
	}
	if s.OpenProbe != nil || s.Identity != committedPinsIdentity(f.rec.Move.Inputs) {
		t.Fatalf("the committed start: probe %v, identity %+v", s.OpenProbe != nil, s.Identity)
	}
	cmd := s.Build()
	for _, kv := range cmd.Env {
		if len(kv) >= len("CARLOS_MOVE_PROBE=") && kv[:len("CARLOS_MOVE_PROBE=")] == "CARLOS_MOVE_PROBE=" {
			t.Fatalf("the committed wake carries %s", kv)
		}
	}
	if err := s.Reserve(t.Context(), 0); err != nil {
		t.Fatal(err)
	}
	j, err := moves.ReadJournal(f.dataDir, poHost, poOp)
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := j.Steps[probeStepStart+"committed-"+strconv.FormatUint(ah.Session, 10)]; !ok {
		t.Fatalf("the committed start is not in the moves attempt journal: %v", j.Steps)
	}
}

// mwDeclare turns f into the landing whose target uk-box is lost, and
// declares it through the console's real path: the record and the claim in
// the deployment store (the target admitted session 20 into the move id
// after the cut at 15, whose seal did not cover the source's replica), a
// signed roster naming both boxes, the move's admission registration, then
// hostops' InspectMoveLoss and DeclareMoveLoss, which run hostresolve's
// declaration with the real moves loss owner. It returns the root's public
// key, the declaration's id and the rollback id the declaration chose.
func mwDeclare(t *testing.T, f *rcFixture) (ed25519.PublicKey, string, string) {
	t.Helper()
	ctx := t.Context()
	pub, priv, err := ed25519.GenerateKey(nil)
	if err != nil {
		t.Fatal(err)
	}
	signer := release.RootSigner(priv)
	clock := time.Now().UTC().Truncate(time.Second)
	control, err := hostops.NewControlWriter(f.claims, pub, func() time.Time { return clock })
	if err != nil {
		t.Fatal(err)
	}
	const bucket = "control"
	d := hostactors.Directory{Store: f.claims, PublicKey: pub, Bucket: bucket}
	if err := d.Initialize(ctx, signer); err != nil {
		t.Fatal(err)
	}
	ro, err := d.Read(ctx)
	if err != nil {
		t.Fatal(err)
	}
	r := hostactors.Clone(ro.Roster)
	r.Boxes["ie-box"] = hostactors.Box{Enrollment: strings.Repeat("a", 32), Namespace: "platform", BoxID: "ie-box", Label: "ie-label", Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-ie"}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}
	r.Boxes["uk-box"] = hostactors.Box{Enrollment: strings.Repeat("b", 32), Namespace: "platform", BoxID: "uk-box", Label: "uk-label", Region: "uk", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-2", InstanceID: "i-uk"}, EvidenceSHA256: strings.Repeat("d", 64), State: "active"}
	if _, err := d.Replace(ctx, signer, ro, r); err != nil {
		t.Fatal(err)
	}
	at := clock.Add(-time.Hour)
	seal := rcCutSeal(false)
	c := f.recoveryClaim()
	c.Holder.Acquisition = "acq"
	c.Holder.Stamp = 18
	c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: poOp, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"},
		Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box", "app_id": "canonical", "admission_id": poOp, "source_enrollment": strings.Repeat("a", 32), "target_enrollment": strings.Repeat("b", 32)},
		Journal: map[string]string{hostclaim.JournalSeal: seal.Encode(), hostclaim.JournalCutSession: "15", hostclaim.JournalSourceStamp: "15", hostclaim.JournalSrcGen: "src",
			hostclaim.JournalEvidenceSource: moves.StoppedKey(moves.ReportPrefix("bab", "jam", poHost, poOp), "source")}}
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	f.putClaim(c)
	flipped, deadline := at, at.Add(2*time.Hour)
	rec := provision.Record{Schema: 1, Host: poHost, Slug: "jam", Account: "bab", App: "jam", Acquisition: "acq", Channel: "edge", Region: "uk", CreatedAt: c.Holder.CreatedAt,
		Move: &provision.MoveState{ID: poOp, Incarnation: f.inc, From: "ie", To: "uk", Cut: "auto", State: provision.MoveLanding, CutPhase: provision.CutFlipped,
			FlippedAt: &flipped, LandingDeadline: &deadline, LandingPhase: provision.LandProbing, Inputs: mpPins(), Source: f.rec.Move.Source}}
	raw, _ := json.Marshal(rec)
	if _, err := f.claims.Put(ctx, provision.RecordKey("bab", "jam", poHost), raw, ""); err != nil {
		t.Fatal(err)
	}
	owner := hostops.RegistrationOwner{Console: "console", Enrollment: strings.Repeat("e", 32), Controller: "coordinator"}
	adm := hostops.AppAdmission{V: 1, AppID: "canonical", Epoch: 1, Revision: 1, State: "open", Effects: map[string]hostops.Effect{},
		Registrations: map[string]hostops.Registration{poOp: {Kind: hostops.RegistrationMove, Host: poHost, Owner: owner, AppID: "canonical", Epoch: 1, Claim: hostops.SavedClaim{ETag: "serving", Gen: 1}, Expires: clock.Add(hostops.RegistrationTTL), CreatedAt: clock, Incarnation: f.inc}}}
	key, err := hostops.AdmissionKey("bab", "jam")
	if err != nil {
		t.Fatal(err)
	}
	raw, _ = json.Marshal(adm)
	if _, err := f.claims.Put(ctx, key, raw, ""); err != nil {
		t.Fatal(err)
	}
	in, err := control.InspectMoveLoss(ctx, signer, bucket, "operator", "bab", "jam", poHost, "uk-label")
	if err != nil {
		t.Fatal(err)
	}
	clock = clock.Add(2 * time.Second)
	sub := hostfence.Submission{Challenge: in.Challenge, Confirmation: in.Details.ID,
		Observation: hostfence.CloudObservation{ObservedAt: in.Details.IssuedAt.Add(time.Second), Account: "123", Region: "eu-west-2", InstanceID: "i-uk", State: "stopped", RequestID: "request"},
		Attestation: hostfence.OperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true, Disks: in.Details.Consequences.DestroyedStorageAssertion}}
	declared, err := control.DeclareMoveLoss(ctx, signer, bucket, "operator", "bab", "jam", poHost, "uk-box", sub)
	if err != nil {
		t.Fatal(err)
	}
	return pub, in.Details.ID, declared.Claim.Op.Intent["rollback"]
}

// TestLostTargetRecoveryRunsAgainstARealDeclaration (Q12; Plan 4 Task
// 10): the source's recovery runs against the claim and the signed loss
// record a real declaration produced (hostresolve with the real moves loss
// owner, through the console's typed writer), never a hand-written one.
// The source's return then takes the claim back (ReturnToSource, probing)
// and its record to probing with the recorded inputs, as the source pass
// does, and the registered source's owners, reading the record from the
// store, admit through the claim's writer with the moves loss owner: the
// declaration is consumed, the target's session is declared lost at 0 (the
// restore does not contain the move id's generation) and the uncovered cut
// at the seal's maximum; the source restores its own baseline.
func TestLostTargetRecoveryRunsAgainstARealDeclaration(t *testing.T) {
	ctx := t.Context()
	f := newRCFixture(t, rcClean)
	pub, lossID, rb := mwDeclare(t, f)
	o := f.read()
	if o.Claim.Op.Kind != hostclaim.KindRestoring || o.Claim.Op.Phase != hostclaim.PhaseReleased || !provision.ValidRollbackID(rb) || !hostclaim.LostTargetReleased(o.Claim, lossID) {
		t.Fatalf("the declaration's claim: %+v", o.Claim.Op)
	}
	loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, lossID)
	if err != nil || loss.Record.State != "committed" {
		t.Fatalf("the loss record: %+v %v", loss.Record, err)
	}
	wantSources := []hostclaim.WriteSource{
		{Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp},
		{Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: "seal:" + poOp},
	}
	if !reflect.DeepEqual(loss.Record.WriteSources, wantSources) {
		t.Fatalf("the declaration froze %+v", loss.Record.WriteSources)
	}

	// The source's pass: the claim back to this box at probing, the record
	// through the clean branch to probing, the inputs it renders recorded.
	if _, err := hostclaim.Update(ctx, f.claims, o, hostclaim.ReturnToSource(poOp, "ie-box", true, hostclaim.ReturnFence{})); err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(ctx, f.claims, f.read(), hostclaim.SetRestoringPhase(poOp, hostclaim.PhaseClaimed, hostclaim.PhaseProbing)); err != nil {
		t.Fatal(err)
	}
	key, err := moves.ReportKeyFromSecrets(mwSecretsKey)
	if err != nil {
		t.Fatal(err)
	}
	inputs := *f.inputs
	inputs.EnvReport = localEnvDigest(key, f.launch.Env)
	*f.inputs = inputs
	g := provision.MoveGuard{ID: poOp, Incarnation: f.inc}
	for _, step := range [][2]string{{provision.RbSourceWaiting, provision.RbSourceClaimed}, {provision.RbSourceClaimed, provision.RbSourceRestoring}, {provision.RbSourceRestoring, provision.RbSourceVerified}, {provision.RbSourceVerified, provision.RbSourceProbing}} {
		if _, err := provision.SetRecordMove(ctx, f.claims, "bab", "jam", poHost, provision.MoveRollbackSourcePhase(g, step[0], step[1], nil)); err != nil {
			t.Fatalf("%s to %s: %v", step[0], step[1], err)
		}
	}
	if _, err := provision.SetRecordMove(ctx, f.claims, "bab", "jam", poHost, provision.MoveRecordRecoveryInputs(g, rb, inputs)); err != nil {
		t.Fatal(err)
	}
	f.route.HoldOp = rb

	// The registered source over the store's records.
	f.owners.Records = MovesRecordsFrom(f.claims)
	w := mwRegister(t, f.poFixture, "ie-box", "ie", f.c8)
	w.srcSnapshot("the source baseline")
	src := w.sourceInstallation()
	w.build(pub, src.inventory, src.registry)
	o = f.read()
	if err := w.wake.stopPredecessor(ctx, o); err != nil {
		t.Fatalf("the declaration as the predecessor's authority: %v", err)
	}
	admitted, err := w.admit(o, f.policy())
	if err != nil {
		t.Fatal(err)
	}
	h := admitted.Claim.Holder
	a := h.Admissions[len(h.Admissions)-1]
	if h.WriterBox != "ie-box" || a.Coverage != "" || a.LossID != lossID || !hostclaim.OperationLossConsumed(admitted.Claim, lossID) {
		t.Fatalf("the loss admission: holder %+v, last %+v", h, a)
	}
	if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) ||
		!reflect.DeepEqual(a.AdditionalWritesLost, []hostclaim.WritesLost{{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}) {
		t.Fatalf("the declared writes: %+v, additional %+v", a.WritesLost, a.AdditionalWritesLost)
	}
	w.restore(admitted, f.policy())
	if got := mwReadDB(t, f.route.DBPath); got != "the source baseline" {
		t.Fatalf("restored %q", got)
	}
}