package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"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"
	"github.com/carlosframework/platform/internal/release"
)

// The target's staging park after the commit and the close (moves spec §7.9
// steps 4 and 5; Plan 5 Task 3). The fixture is the source's finalization
// harness (newFinalHarness: a landing driven to restored, the source box ie
// with its real registry row and artifacts), given a probe so that the
// TARGET box (uk) probes, commits and opens through Plan 3's committed arm,
// not by hand. Both boxes run the real park engine over the fake versioned
// buckets with one fake lease table. Nothing in production runs the
// source region's acknowledgement yet (C4's regional runner), so
// regionalAck stands in for it.

func newCloseHarness(t *testing.T, opts ...p1Option) *finalHarness {
	t.Helper()
	h := newFinalHarness(t, opts...)
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
		rec.Move.CutAttempts = 2
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	p := &probeLog{t: t, h: h.landingHarness}
	h.probe, h.probeLive = p.wake, p.liveProbe
	h.ownAuthorize = true
	return h
}

// chore finds the op's park-replica in region: the staging lane when lane
// is true, the location otherwise.
func (h *finalHarness) chore(t *testing.T, region string, lane bool) *hostclaim.Obligation {
	t.Helper()
	for _, o := range h.claim(t).Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Region == region && o.Park != nil && isStagingLocation(o.Park.Location) == lane {
			return &o
		}
	}
	return nil
}

// regionalAck is C4's regional runner for the source region: the park
// engine over data-ie, run once at the harness's clock.
func (h *finalHarness) regionalAck(t *testing.T, o *hostclaim.Obligation) error {
	t.Helper()
	src, dst := park.PlacementRegion(h.src), park.PlacementRegion(h.dst)
	e := park.Engine{Store: h.vs, Claims: h.s, Bucket: h.srcBucket, Region: src, Leases: h.leases, Now: h.clock.Now,
		NewID: obligationID, View: sourceTestViewFor(src, dst), Authorize: sourceTestAuthorizeFor(h.srcBox, src)}
	return e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID})
}

// clockTo moves the harness's clock to at, never back.
func (h *finalHarness) clockTo(t *testing.T, at time.Time) {
	t.Helper()
	if d := at.Sub(h.clock.now); d > 0 {
		h.clock.Advance(d)
	}
	if h.clock.now.After(at.Add(time.Second)) {
		t.Fatalf("fixture: the clock is already at %v, past %v", h.clock.now, at)
	}
}

// clockPast moves the harness's clock past at, if it is not there already.
func (h *finalHarness) clockPast(at time.Time) {
	if d := at.Sub(h.clock.now); d >= 0 {
		h.clock.Advance(d + time.Second)
	}
}

// committedAndOpen asserts the target committed and opened the landing.
func (h *finalHarness) committedAndOpen(t *testing.T) {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m == nil || m.State != provision.MoveCommitted || m.OpenedAt == nil {
		t.Fatalf("the target did not commit and open: %+v", m)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
		t.Fatalf("the claim: %+v", c.Op)
	}
}

// notClosed asserts the move is still open on both halves.
func (h *finalHarness) notClosed(t *testing.T, why string) {
	t.Helper()
	if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveCommitted {
		t.Fatalf("%s: the record closed: %+v", why, m)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
		t.Fatalf("%s: the claim was released: %+v", why, c.Op)
	}
}

// done asserts the close: the claim serving under the target with this
// move recorded as released and nothing owed, the record at rest in uk with
// LastMove done.
func (h *finalHarness) done(t *testing.T) {
	t.Helper()
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.Region != h.dst || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "done" {
		t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != h.dstBox || hostclaim.RequireReleasedMove(p1ID)(c) != nil || len(c.Obligations) != 0 {
		t.Fatalf("the claim after the close: %+v obligations %+v", c.Op, c.Obligations)
	}
}

// TestStagingParkAcknowledgesAtSealBoundPlusH (Review Focus 1, witness
// wStagingScopeDeadline; §7.9 step 4, K6): the commit queues the lane's
// park scoped to the seal's bound T, and the target runs it through
// laneParkTarget. At T + 16 min the run is ErrPending and nothing is
// acknowledged; a restarted worker at T + 17 min + 1 s acknowledges, and
// the scope it acknowledged is still T's (quiet_after = T + H).
func TestStagingParkAcknowledgesAtSealBoundPlusH(t *testing.T) {
	t.Parallel()
	h := newCloseHarness(t)
	h.land(t)
	h.committedAndOpen(t)
	seal, err := journalledSeal(h.claim(t))
	if err != nil {
		t.Fatal(err)
	}
	bound := seal.Bound()
	o := h.chore(t, "uk", true)
	if o == nil {
		t.Fatal("the commit queued no staging park")
	}
	if want := moves.StagingScope(p1Host, p1ID, h.inc, bound); *o.Park != want || o.Owner != "region:uk" {
		t.Fatalf("the staging park %+v %+v, want %+v", o, o.Park, want)
	}
	// The first pass archived the lane; nothing is acknowledged yet.
	lane := moves.ScopeLocation(p1Host, p1ID) + "/"
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, lane) {
			t.Fatalf("the lane still holds %s after the target's pass", k)
		}
	}
	// T + 16 min: the engine answers ErrPending, and the pass acknowledges
	// nothing.
	h.clockTo(t, bound.Add(16*time.Minute))
	e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now,
		NewID: obligationID, View: standInView("uk", roleTarget, h.read(t, p1Host).Move), Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
	if err := e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); !errors.Is(err, park.ErrPending) {
		t.Fatalf("at T + 16 min the run answered %v, want ErrPending", err)
	}
	h.land(t)
	if again := h.chore(t, "uk", true); again == nil || !again.Park.EvidenceAt.Equal(bound) {
		t.Fatalf("at T + 16 min: %+v", again)
	}
	// A restarted target at T + 17 min + 1 s acknowledges.
	h.freshTarget()
	h.clockTo(t, bound.Add(park.SignatureAge+moves.ParkRequestLimit+time.Second))
	h.land(t)
	if again := h.chore(t, "uk", true); again != nil {
		t.Fatalf("the staging park is still owed past T + H: %+v", again)
	}
	if q := h.claim(t).QuietAfter["uk"]; !q.Equal(bound.Add(park.SignatureAge + moves.ParkRequestLimit)) {
		t.Fatalf("quiet_after %v, want T + H = %v", q, bound.Add(park.SignatureAge+moves.ParkRequestLimit))
	}
	if n := len(h.vs.callsOn("copy-version data-uk/parked/" + h.inc + "/" + p1Host + "/" + o.Retirement + "/moves/")); n == 0 {
		t.Fatal("nothing was archived under the lane's retirement")
	}
}

// TestCloseWaitsForBothObligations (§7.9 step 5): with the target open and
// the source parked, the close still waits while the source region's
// location park is owed; once both parks are acknowledged the claim is
// serving under uk-box and the record at rest in uk.
func TestCloseWaitsForBothObligations(t *testing.T) {
	t.Parallel()
	h := newCloseHarness(t)
	h.land(t)
	h.committedAndOpen(t)
	h.sourcePass(t)
	if m := h.read(t, p1Host).Move; m.SourceParkedAt == nil {
		t.Fatalf("the source did not finalize: %+v", m)
	}
	h.land(t)
	h.notClosed(t, "both parks owed")
	// Past the staging lane's horizon the target acknowledges it; the
	// source's location park is still owed (no regional runner ran it).
	lane := h.chore(t, "uk", true)
	h.clockPast(lane.Park.EvidenceAt.Add(park.SignatureAge + lane.Park.RequestLimit))
	h.land(t)
	if h.chore(t, "uk", true) != nil {
		t.Fatal("the staging park was not acknowledged")
	}
	src := h.chore(t, "ie", false)
	if src == nil {
		t.Fatal("fixture: the source's location park was acknowledged by the finalization")
	}
	h.land(t)
	h.notClosed(t, "the source's location park owed")
	// The source region's runner acknowledges after its horizon; then the
	// target's next pass closes.
	h.clockPast(src.Park.EvidenceAt.Add(park.SignatureAge + src.Park.RequestLimit))
	if err := h.regionalAck(t, src); err != nil {
		t.Fatalf("the source region's acknowledgement: %v", err)
	}
	if h.chore(t, "ie", false) != nil {
		t.Fatal("the source's location park was not acknowledged")
	}
	h.land(t)
	h.done(t)
	// A further pass of either box changes nothing.
	before := h.claim(t).Gen
	h.land(t)
	h.sourcePass(t)
	if after := h.claim(t).Gen; after != before {
		t.Fatalf("a pass after the close wrote the claim (%d to %d)", before, after)
	}
}

// TestCloseIsRecordFirst (§7.5, §19 K8 as Paul's ruling M4 amends them):
// the close clears the record before it releases the claim. A crash between
// leaves the record at rest with LastMove done and the claim
// landing/committed with nothing owed; no new move can start (the claim is
// not serving), and no box pass writes the claim again: the release is the
// console's recovery pass (TestBoxClosesAreRecordFirst; the console's
// TestRecordFirstCloseCrashIsFinishedByThePass). Without a crash the same
// pass releases.
func TestCloseIsRecordFirst(t *testing.T) {
	t.Parallel()
	h := newCloseHarness(t)
	h.land(t)
	h.sourcePass(t)
	h.clock.Advance(24 * time.Hour)
	if err := h.regionalAck(t, h.chore(t, "ie", false)); err != nil {
		t.Fatal(err)
	}
	h.landAt = "after-close-record"
	h.crashLand(t)
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != "done" {
		t.Fatalf("the record did not close first: %+v %+v", rec.Move, rec.LastMove)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
		t.Fatalf("the claim at the crash point: %+v", c.Op)
	}
	if err := hostclaim.StartGuard(c, []string{"uk"}, []string{"uk-box"}, ""); err == nil {
		t.Fatal("a new operation could start over the unreleased claim")
	}
	gen := c.Gen
	h.freshTarget()
	h.land(t)
	if after := h.claim(t).Gen; after != gen {
		t.Fatalf("a box pass after the record's close wrote the claim (%d to %d)", gen, after)
	}

	// Without the crash, the pass that closes the record releases.
	h2 := newCloseHarness(t)
	h2.land(t)
	h2.sourcePass(t)
	h2.clock.Advance(24 * time.Hour)
	if err := h2.regionalAck(t, h2.chore(t, "ie", false)); err != nil {
		t.Fatal(err)
	}
	h2.land(t)
	h2.done(t)
}

// Any participant that sees both stamps and nothing owed closes (§7.9 step
// 5): the source's pass that stamps source_parked_at closes too when the
// target has opened and both parks are already acknowledged.
func TestTheSourceClosesWhenItFinishesLast(t *testing.T) {
	t.Parallel()
	h := newCloseHarness(t)
	h.landFault = func(point string) error {
		if point == "after-close-record" {
			t.Fatal("the target closed before the source parked")
		}
		return nil
	}
	h.land(t)
	h.committedAndOpen(t)
	h.clock.Advance(24 * time.Hour)
	h.land(t)
	if h.chore(t, "uk", true) != nil {
		t.Fatal("the staging park was not acknowledged")
	}
	h.notClosed(t, "the source has not parked")
	// The source's location park is acknowledged in the finalization's own
	// run, past its horizon; the same pass stamps and closes.
	h.sourcePass(t)
	h.done(t)
}

// K9: the console's commit (Task 7a) builds its scopes from the record and
// the claim alone, through moves.CommitScopes with FlippedAt, so whichever
// of the box and the console commits first writes the same obligations,
// byte for byte, and the other's replay is no change.
func TestConsoleCommitMatchesTheBoxs(t *testing.T) {
	t.Parallel()
	h, _ := probedLanding(t)
	var pre *hostclaim.Claim
	h.landFault = func(point string) error {
		switch point {
		case "before-commit":
			pre = h.claim(t)
		case "after-claim-committed":
			return errCrash
		}
		return nil
	}
	h.land(t)
	if pre == nil || pre.Op.Phase != hostclaim.PhaseLandingProbed {
		t.Fatalf("fixture: the claim before the commit: %+v", pre)
	}
	box := h.claim(t)
	if box.Op.Phase != hostclaim.PhaseCommitted {
		t.Fatalf("the box did not commit the claim: %+v", box.Op)
	}
	rec := h.read(t, p1Host)
	source, staging, err := moves.CommitScopes(pre, rec.Move.FlippedAt)
	if err != nil {
		t.Fatal(err)
	}
	var minted []string
	for _, o := range box.Obligations[len(pre.Obligations):] {
		minted = append(minted, o.ID, o.Retirement)
	}
	newID := func() string {
		if len(minted) == 0 {
			t.Fatal("the console's commit minted more ids than the box's")
		}
		id := minted[0]
		minted = minted[1:]
		return id
	}
	console := pre
	if err := hostclaim.Commit(p1ID, rec.Move.From, pre.Op.Intent["source"], rec.Move.To, source, staging, newID)(console); err != nil {
		t.Fatal(err)
	}
	got, _ := json.Marshal(console.Obligations)
	want, _ := json.Marshal(box.Obligations)
	if !bytes.Equal(got, want) || len(minted) != 0 {
		t.Fatalf("the console's commit queues\n%s\nthe box's\n%s", got, want)
	}
	// Without the flip's time there is no source scope (H7).
	if _, _, err := moves.CommitScopes(pre, nil); err == nil {
		t.Fatal("scopes without the flip")
	}
}

// Carried follow-up (d) (Plan 3 Task 8; Task 3 rulings): the finalization's
// moves-owned rename runs inside the host guard with the claim and the
// record read fresh there, so authority that changes before it renames
// nothing at the live paths and journals no plan.
func TestSourceFinalizationRenameIsGuarded(t *testing.T) {
	t.Parallel()
	for name, change := range map[string]func(t *testing.T, h *finalHarness){
		"the claim's executor is this box again": func(t *testing.T, h *finalHarness) {
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "ie-box" })
		},
		"the claim has released the move": func(t *testing.T, h *finalHarness) {
			rewriteClaimOp(t, h.s, func(op *hostclaim.Op) {
				*op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box", Journal: map[string]string{}}
			})
		},
		"the inventory claims a path": func(t *testing.T, h *finalHarness) {
			rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
				if c.Inventory == nil {
					c.Inventory = map[string]hostclaim.InventoryEntry{}
				}
				c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 9}}
			})
		},
		"the record is no longer this commit": func(t *testing.T, h *finalHarness) {
			if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
				rec.Move.ID = p1ID2
				return nil
			}); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			h := newFinalHarness(t)
			h.commit(t)
			changed := false
			inner := h.cfg.finalFault
			h.cfg.finalFault = func(point string) error {
				if point == "after-"+finalStepInstallation && !changed {
					changed = true
					change(t, h)
				}
				return inner(point)
			}
			h.sourcePass(t)
			if !changed {
				t.Fatalf("the rename's guard point was never reached: %v", h.finalHits)
			}
			want := []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}
			if got := h.live(t); strings.Join(got, " ") != strings.Join(want, " ") {
				t.Fatalf("live paths moved: %v, want %v", got, want)
			}
			if got := h.parked(t); len(got) != 0 {
				t.Fatalf("parked: %v", got)
			}
			if j := h.sourceJournal(t); j.Done(finalStepLocal+"-plan") || j.Done(finalStepLocal) {
				t.Fatalf("a plan was journalled: %v", j.Steps)
			}
		})
	}
}

// The close harness's production probe mode (home part 2, Task 2). The
// fake (probeLog.wake) admits with a literal box and region and so never
// meets the production owners, their admission evidence or the C8 files
// publication, which is where a home box's roster name meets the move's
// @home. This mode wakes the target's probe through the production
// movesProbeOwners over the claim runtime (claim Plan 9 Group D's
// ClaimWakeAssembly, as TestMovesProbeOwnersThroughTheClaimRuntime drives
// it), with the target box's identity from the roster: Box the target box,
// Region its claim engine region (its roster name, ie for a home box),
// HomeRegion and FleetHomeRegion the fleet's designation. Admit, Prepare,
// the probe's open, the attempt's reservation and Materialize are the
// claim's and the owners' own. What stands in is what main cannot run here:
// the launch (every claim start is refused on main until the isolated root
// launcher lands), the held works and render (the claim's move read access
// over this box's registry), and the activator's live probe entry. A
// Materialize that passes is the attempt passing.

// p1Work is a held action's selected work in the production probe mode:
// it runs fn, and records the identities it ran under.
type p1Work struct {
	order *[]string
	name  string
}

func (w p1Work) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
	*w.order = append(*w.order, w.name+":"+id.Config+"|"+id.Release)
	return fn()
}

// p1Markers is the claim's marker store over the harness's fake versioned
// buckets: a conditional put on the current version's etag.
type p1Markers struct{ vs *fakeVersioned }

func (m p1Markers) PutMarker(ctx context.Context, bucket, key string, body []byte, expected string, check func(context.Context) error) (string, error) {
	if err := check(ctx); err != nil {
		return "", err
	}
	current := ""
	if o, ok := m.vs.f.buckets[bucket][key]; ok {
		current = o.ETag
	}
	if current != expected {
		return "", hostbox.ErrChanged
	}
	ver := m.vs.f.nextVersion()
	m.vs.f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
	m.vs.f.setVersion(bucket, key, ver)
	return ver, nil
}

// p1FilesStore is the target's data bucket as a release.Store for the C8
// provider (movesFiles.Files): current reads with the version as the etag,
// create-only and compare-and-swap writes.
type p1FilesStore struct {
	vs     *fakeVersioned
	bucket string
}

func (s p1FilesStore) Get(_ context.Context, key string) ([]byte, string, error) {
	o, ok := s.vs.f.buckets[s.bucket][key]
	if !ok {
		return nil, "", os.ErrNotExist
	}
	return []byte(o.body), s.vs.versionID(s.bucket, key), nil
}

func (s p1FilesStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if ifMatch == "" {
		ver, err := s.vs.PutIfAbsent(ctx, s.bucket, key, body)
		if errors.Is(err, lease.ErrExists) {
			return "", release.ErrCASConflict
		}
		return ver, err
	}
	if _, ok := s.vs.f.buckets[s.bucket][key]; !ok || s.vs.versionID(s.bucket, key) != ifMatch {
		return "", release.ErrCASConflict
	}
	ver := s.vs.f.nextVersion()
	s.vs.f.store(s.bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
	s.vs.f.setVersion(s.bucket, key, ver)
	return ver, nil
}

func (s p1FilesStore) List(_ context.Context, prefix string) ([]string, error) {
	var out []string
	for k := range s.vs.f.buckets[s.bucket] {
		if strings.HasPrefix(k, prefix) {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out, nil
}

func (p1FilesStore) ListDirs(context.Context, string) ([]string, error) {
	return nil, errors.New("test: not implemented")
}

func (p1FilesStore) Delete(context.Context, string) error {
	return errors.New("test: not implemented")
}

// prodProbe is the production mode's state on one box: the claim runtime
// (claim Plan 9 Group D's ClaimWakeAssembly, the recovery fixture's
// controller) with the moves owners bound, on the box the side names.
type prodProbe struct {
	t      *testing.T
	h      *phase1Harness
	side   prodSide
	ctl    *recoveryController
	owners *movesProbeOwners
	// order is the held works, render and release, in the order they ran.
	order []string
	// evidence counts the writer's admission evidence calls, and
	// rejectEvidence, when set, refuses them before the owners are asked
	// (the rejected-evidence witness).
	evidence       int
	rejectEvidence error
	// writerEvidence is the writer's AdmissionEvidence as bound.
	writerEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
	// selected is the probe's launch inputs as the owners select them for
	// the attempt, and attempt the reservation the runtime journalled for
	// it: root's launch check compares the reservation's LaunchInputs with
	// the hash of the launch it is asked to start.
	selected ClaimLaunchInputs
	attempt  hostclaim.WakeRetryAttempt
	// admissions are the admissions the runtime made, one per attempt, and
	// admitted the claim each left, the last attempt's last; observed is
	// the claim each start read before its admission.
	admissions []hostclaim.AdmissionRecord
	admitted   hostclaim.Observed
	observed   []hostclaim.Observed
	// stopped counts the predecessor writer sessions this box's own wakes
	// stopped through the production Stop.
	stopped int
	// materialized is the claim and the launch inputs the runtime last
	// materialized: the admitted inputs, which for a fresh generation are
	// not the selected ones (each Select mints its own generation).
	materialized   hostclaim.Observed
	materializedIn ClaimLaunchInputs
	// afterMaterialize, when set, runs in the n-th wake (from 1) after
	// Materialize and before the launch, while the attempt is live; an error
	// fails the attempt there (the runtime's Failed), as an attempt that
	// failed before readiness, or an agent that died there, leaves it (home
	// part 2, Tasks 3 and 4).
	afterMaterialize func(n int) error
	// beforeAdmit, when set, sees each start's claim and the owners'
	// selection before the runtime's admission.
	beforeAdmit func(o hostclaim.Observed, in ClaimLaunchInputs)
	wakes       int
	last        activator.ProbeResult
	live        bool
	seq         int
}

// prodSide is the box a production mode runs on: its roster box, its
// placement code (its composer's region), its data directory, data bucket
// and the bucket's AWS region, its registry row, and the inputs its held
// render renders (the move's pins for a landing, the recorded recovery
// inputs for a recovery).
type prodSide struct {
	box, code, dataDir, bucket, aws string
	route                           func() registry.Route
	inputs                          func() *provision.MoveInputs
}

// productionProbe switches the harness's probe to the production mode: the
// target's probe wakes through movesProbeOwners over the claim runtime, with
// the target box's roster identity.
func (h *landingHarness) productionProbe(t *testing.T) *prodProbe {
	t.Helper()
	p := newProdWake(t, h.phase1Harness, prodSide{box: h.dstBox, code: h.dst, dataDir: h.tdir, bucket: h.dstBucket, aws: h.dstAWS,
		route:  func() registry.Route { return h.route },
		inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.Inputs }})
	h.probe, h.probeLive = p.wake, p.liveProbe
	return p
}

// newProdWake builds the production mode on side's box. The owners are the
// production movesProbeOwners (which dispatch a recovery op to their
// recovery owners) with the real C8 provider, bound into the claim runtime
// with the owners' admission evidence, the moves attempt journal and the
// production artifact retirement (claimArtifactRetirement over park.Local,
// as claimconstruction.go wires RetireArtifacts and RecoverArtifacts).
// What stands in is what main cannot run here: the launch (every claim
// start is refused on main until the isolated root launcher lands), its
// readiness, the held works and render (the claim's move read access over
// this box's registry), root's stop transport (the recovery fixture's
// coordinator), the cross-box cold handover (AfterAdmission) and the
// activator's live-probe entry. A Materialize that passes is the attempt
// passing.
func newProdWake(t *testing.T, h *phase1Harness, side prodSide) *prodProbe {
	t.Helper()
	if !h.realEnv {
		// The probe's launch is the box's own composition; a source that
		// pinned the h.env fixture would never match it.
		t.Fatal("fixture: the production probe mode needs realLaunchEnv")
	}
	// Claim Plan 15: admissions through the real CAS record their origin
	// under both switches; the real C8's proof reads that record.
	t.Setenv("CARLOS_HOST_CLAIM", "1")
	t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
	actor := h.roster[side.box]
	journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { journal.Close() })
	p := &prodProbe{t: t, h: h, side: side}
	guard := hostguard.Guard{DataDir: side.dataDir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}
	f := &recoveryFixture{t: t, claims: h.s, evidence: release.NewDirStore(t.TempDir()), signer: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)),
		guard: guard, journal: journal, kernel: &recoveryKernel{journal: journal}, now: h.clock.Now(), recordPrelaunch: true,
		data: h.vs, bucket: side.bucket, box: actor.Box, enrollment: actor.Enrollment, region: actor.Region, leaseRegion: side.aws}
	key, err := moves.ReportKeyFromSecrets(h.cfg.SecretsKey)
	if err != nil {
		t.Fatal(err)
	}
	filesDir := func(host string) (string, string) {
		return filepath.Join(side.dataDir, host), filepath.Join(side.dataDir, host+".env")
	}
	owners := &movesProbeOwners{HomeRegion: actor.HomeRegion, FleetHomeRegion: actor.FleetHomeRegion, Claims: h.s, Records: MovesRecordsFrom(h.s),
		ReadyPath: movesReadyPath(h.s), Data: h.vs, Bucket: side.bucket, Region: actor.Region, Box: actor.Box, DataDir: side.dataDir, FilesDir: filesDir,
		Markers: p1Markers{h.vs},
		Route:   func(host string) (*registry.Route, error) { rt := side.route(); return &rt, nil },
		// The held row's launch: what this box would launch, composed by
		// the production composer over its data directory with its own
		// region, as claimMoveLaunchSource composes it
		// (ExecBackend.LaunchLayers, CARLOS_MOVE_PROBE dropped), the live
		// release, and the env report made from that same composition, whose
		// digest is the launch's move-portable pin.
		Launch: func(_ context.Context, rt registry.Route) (ClaimServingLaunch, moves.EnvReport, error) {
			composed, supplied, _, err := (&activator.ExecBackend{DataDir: side.dataDir, Region: side.code}).LaunchLayers(rt)
			if err != nil {
				return ClaimServingLaunch{}, moves.EnvReport{}, err
			}
			delete(composed, "CARLOS_MOVE_PROBE")
			env := make([]string, 0, len(composed))
			for k, v := range composed {
				env = append(env, k+"="+v)
			}
			sort.Strings(env)
			return ClaimServingLaunch{Executable: "/opt/carlos/live/" + p1Host, ExecutableSHA256: strings.Repeat("a", 64), Env: env, Version: "v7"}, moves.ReportLaunch(key, composed, supplied), nil
		},
		EnvKey:   key,
		Admitted: MovesRecordAdmittedAttempts,
		Works: func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
			p.order = append(p.order, "select:"+action)
			return p1Work{order: &p.order, name: action}, nil
		},
		Render: func(ctx context.Context, _ registry.Route, _ string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) {
			p.order = append(p.order, "render")
			in := side.inputs()
			hr := appconfig.HeldRender{BundleETags: in.BundleETags, OverlayETag: in.OverlayETag}
			if _, err := work(hr); err != nil {
				return appconfig.HeldRender{}, err
			}
			return hr, nil
		},
		Release: func(ctx context.Context, _ registry.Route, version string, work hostwrite.Work) error {
			p.order = append(p.order, "release:"+version)
			return work.With(ctx, hostwrite.Identity{Release: version}, func() error { return nil })
		},
		NewID: func() string { p.seq++; return fmt.Sprintf("fresh%d", p.seq) },
		Now:   h.clock.Now}
	// The C8 provider is the real one over the box's bucket, as the claim
	// assembly binds it, with Plan 15's proof over the claim store.
	owners.Files = &movesFiles{Files: p1FilesStore{vs: h.vs, bucket: side.bucket}, Data: h.vs, Bucket: side.bucket, HomeRegion: actor.FleetHomeRegion, Region: actor.Region, Box: actor.Box,
		DataDir: side.dataDir, EnvFile: envFileOf(filesDir), Proof: ClaimOriginProof{Claims: h.s, HomeRegion: actor.FleetHomeRegion, Region: actor.Region}}
	if !owners.valid() {
		t.Fatal("fixture: the production probe owners are not valid")
	}
	attempts := &movesProbeAttempts{Claims: h.s, Records: MovesRecordsFrom(h.s), DataDir: side.dataDir, NewID: owners.NewID, Now: h.clock.Now, Admitted: MovesRecordAdmittedAttempts}
	bound := owners.MovesOwners(attempts)
	dispatch := ClaimOwnerDispatch{Moves: &bound.ClaimKindOwners}
	// The writer's admission evidence is the owner dispatch's, as the claim
	// assembly binds it (claimassembly.go), bound before the runtime is
	// built: Build captures the writer, so a later swap would leave the
	// controller's allow-all evidence in force.
	f.homeRegion, f.fleetHomeRegion = actor.HomeRegion, actor.FleetHomeRegion
	f.admissionEvidence = func(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, plan *genselect.Plan) error {
		p.evidence++
		if p.rejectEvidence != nil {
			return p.rejectEvidence
		}
		return dispatch.AdmissionEvidence(ctx, o, in, plan)
	}
	p.writerEvidence = f.admissionEvidence
	c := f.controller("boot-1", "controller-1", 4101, "start-1")
	c.wake.Inputs = p1RecordingInputs{ClaimOwnerDispatch: dispatch, p: p}
	c.wake.Continue = hostclaim.SessionContinues
	c.wake.Operations = bound.Operations
	c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
	// The production artifact retirement: a restore over a database of
	// another generation retires that installation through park.Local's
	// RetireSuperseded, which re-tags its inventory paths with the admitted
	// installation's stamp.
	retirement := claimArtifactRetirement{Local: park.Local{Guard: guard, Claims: h.s, DataDir: side.dataDir, Box: actor.Box}, Continue: hostclaim.SessionContinues}
	c.wake.RetireArtifacts, c.wake.RecoverArtifacts = retirement.Retire, retirement.Recover
	p.ctl, p.owners = c, owners
	return p
}

// p1RecordingInputs is the owner dispatch as the runtime's inputs, every
// method its own, with Materialize recording what the runtime admitted.
type p1RecordingInputs struct {
	ClaimOwnerDispatch
	p *prodProbe
}

func (r p1RecordingInputs) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	r.p.materialized, r.p.materializedIn = o, in
	return r.ClaimOwnerDispatch.Materialize(ctx, o, in, work)
}

// wake is one probe attempt through the claim runtime: the activator's
// ProbeWake with its start policy, as far as the claim runs on main.
func (p *prodProbe) wake(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) {
	policy := activator.StartPolicy{MaxStarts: 5, Window: time.Duration(moves.ProbeSeconds * float64(time.Second)), Probe: true,
		Deadline: cfg.Deadline, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
	return p.start(ctx, p.side.route(), policy)
}

// recoveryWake is the activator's recovery probe entry
// (ClaimRecoveryProbeWake) on this box: one claim wake under the
// adapter's policy.
func (p *prodProbe) recoveryWake(ctx context.Context, rt registry.Route, policy activator.StartPolicy) error {
	_, err := p.start(ctx, rt, policy)
	return err
}

// start is one claim wake's first start, in the activator's order
// (claimwake.go): Prerequisites' stop of this box's own predecessor writer
// session (the production Stop, through the fixture's root coordinator),
// Admit, Prepare, Restore (the production cold restore of the selected
// baseline, retiring a superseded installation), then the attempt's
// Reserve and Materialize; a failure from then on is the attempt's
// (Failed). The launch, its readiness and Healthy stand in: a Materialize
// that passes is the attempt passing.
func (p *prodProbe) start(ctx context.Context, rt registry.Route, policy activator.StartPolicy) (activator.ProbeResult, error) {
	p.live = false
	h := p.h
	o, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	if prior := o.Claim.Holder; prior != nil && prior.WriterBox == p.side.box && prior.Session != 0 {
		if err := p.ctl.wake.Stop(ctx, o); err != nil {
			return activator.ProbeResult{}, fmt.Errorf("stop the predecessor session %d: %w", prior.Session, err)
		}
		p.stopped++
		if o, err = hostclaim.Read(ctx, h.s, p1Host); err != nil {
			return activator.ProbeResult{}, err
		}
	}
	live, cancel := context.WithCancel(context.Background())
	p.t.Cleanup(cancel)
	done := make(chan struct{})
	go func() { <-live.Done(); close(done) }()
	p.ctl.wake.mu.Lock()
	p.ctl.wake.pending[p1Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
	p.ctl.wake.mu.Unlock()
	p.observed = append(p.observed, o)
	if p.selected, err = p.owners.Select(ctx, rt, o, policy); err != nil {
		return activator.ProbeResult{}, err
	}
	if p.beforeAdmit != nil {
		p.beforeAdmit(o, p.selected)
	}
	admitted, err := p.ctl.runtime.Admit(ctx, rt, o, policy)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	ah := admitted.Claim.Holder
	p.admissions = append(p.admissions, ah.Admissions[len(ah.Admissions)-1])
	p.admitted = admitted
	p.wakes++
	s, err := p.ctl.runtime.Prepare(ctx, rt, admitted, policy)
	if err != nil {
		return activator.ProbeResult{}, err
	}
	if err := s.Restore(ctx); err != nil {
		return activator.ProbeResult{}, fmt.Errorf("restore: %w", err)
	}
	if err := s.Reserve(ctx, 0); err != nil {
		return activator.ProbeResult{}, err
	}
	p.attempt = s.CurrentAttempt()
	if err := s.Materialize(ctx); err != nil {
		return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0))
	}
	if s.OpenProbe != nil {
		// At the exec (the start's beforeStart hook).
		if _, err := s.OpenProbe(ctx, h.clock.Now()); err != nil {
			return activator.ProbeResult{}, err
		}
	}
	if p.afterMaterialize != nil {
		if err := p.afterMaterialize(p.wakes); err != nil {
			return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0))
		}
	}
	p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation}
	p.live = true
	return p.last, nil
}

// liveProbe is the activator's LiveProbe over the mode's last passing call.
func (p *prodProbe) liveProbe(host string) (string, activator.ProbeResult, bool) {
	if host != p1Host || !p.live {
		return "", activator.ProbeResult{}, false
	}
	return p1ID, p.last, true
}