package host

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

	"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/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/rootstate"
)

// Task 7b: the root-side held release under the host claim (moves after
// claim Plan 12; task-7b-design.md and platform-1e's binding review of
// 2026-10-12). The sandboxed agent cannot write /opt/carlos, so under the
// claim it asks root, over the root control socket, to stage T7's release
// (never publishing it) and to publish the held row's release at configured
// and before a probe's start. Root re-reads the claim, the record, the
// registry row and the render-hash journal, refuses anything that does not
// match before any effect, and publishes only inside its own held-release
// selection.

const (
	hrRendered = "sha256:" + "aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11"
	hrOther    = "sha256:" + "bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22"
	hrRB       = "rb_abcdefghijklmnopqrstuvwxyz"
)

var hrPeer = StopPeer{UID: 1000, PID: 4242, ProcessStart: "77", BootID: "boot", ControllerID: "edge"}

// hrSelector is root's held-release selection as the tests see it: each
// selection is recorded with the claim tuple it was made at, and each write
// it is handed runs, recording the identity it names.
type hrSelector struct {
	mu       sync.Mutex
	f        *hrFixture
	selected []string
	ids      []hostwrite.Identity
	refuse   error
}

type hrWork struct {
	s   *hrSelector
	sel hostbox.Selection
}

func (w hrWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
	w.s.mu.Lock()
	w.s.ids = append(w.s.ids, id)
	w.s.mu.Unlock()
	return fn()
}

func (s *hrSelector) count() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.selected)
}

// hrFixture is one box (uk-box in uk) with a held row of bab/app at a
// landing's configured step: the claim, the record, the registry row, the
// render journal, a store with v7 shipped and authorized, and root.
type hrFixture struct {
	s                   release.Store
	opt, data, receipts string
	inc                 string
	mu                  sync.Mutex
	claim               hostclaim.Claim
	rec                 provision.Record
	row                 *registry.Route
	sel                 *hrSelector
	claimReads          int
	root                *movesHeldReleaseRoot
}

func newHRFixture(t *testing.T) *hrFixture {
	t.Helper()
	s := release.NewDirStore(t.TempDir())
	shipPinnedV7(t, s)
	inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created)
	if err != nil {
		t.Fatal(err)
	}
	f := &hrFixture{s: s, opt: t.TempDir(), data: t.TempDir(), receipts: t.TempDir(), inc: inc}
	f.claim = hostclaim.Claim{V: 1, Host: p1Host, Gen: 5,
		Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "app", p1Host), Account: "bab", App: "app", Incarnation: inc,
			CreatedAt: p1Created, Location: "instances/" + p1Host + "/" + inc, Generation: p1ID, Stamp: 7, Session: 3, WriterBox: "uk-box"},
		Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: p1ID, Phase: provision.LandConfigured, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"},
			Intent: map[string]string{"source": "ie-box", "target": "uk-box", "to": "uk"}}}
	f.rec = provision.Record{Schema: 1, Host: p1Host, Slug: "s", App: "app", Account: "bab", Region: "uk", CreatedAt: p1Created,
		Move: &provision.MoveState{ID: p1ID, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandConfigured,
			Inputs: &provision.MoveInputs{Release: "v7"}}}
	f.row = &registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "landing", HoldOp: p1ID}
	f.sel = &hrSelector{f: f}
	raw, err := json.Marshal(moves.SourceReport{V: 1, Box: "ie-box", MoveID: p1ID, At: time.Now(), Release: "v7"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), moves.SourceReportKey(moves.ReportPrefix("bab", "app", p1Host, p1ID)), raw, ""); err != nil {
		t.Fatal(err)
	}
	f.journal(t, p1ID, hrRendered)
	f.root = &movesHeldReleaseRoot{
		OpenRelease:  func() (rootstate.Store, error) { return &plan14ReleaseStore{}, nil },
		Authenticate: nil,
		Actor: func(context.Context) (hostbox.Actor, error) {
			return hostbox.Actor{Box: "uk-box", Region: "uk", Eligible: true}, nil
		},
		Store:   s,
		Claims:  f.claims,
		Records: f.records,
		Route: func(host string) (*registry.Route, error) {
			f.mu.Lock()
			defer f.mu.Unlock()
			if f.row == nil || host != f.row.Host {
				return nil, nil
			}
			rt := *f.row
			return &rt, nil
		},
		Works: func(_ context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error) {
			f.sel.mu.Lock()
			defer f.sel.mu.Unlock()
			f.sel.selected = append(f.sel.selected, host+"/"+action)
			if f.sel.refuse != nil {
				return nil, f.sel.refuse
			}
			f.mu.Lock()
			c := f.claim
			f.mu.Unlock()
			return hrWork{s: f.sel, sel: hostbox.Selection{Host: host, Action: action, Box: actor.Box, Region: actor.Region,
				Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp}, Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase,
				Generation: c.Holder.Generation, Session: c.Holder.Session}}, nil
		},
		Selected: func(w hostwrite.Work) (hostbox.Selection, error) {
			hw, ok := w.(hrWork)
			if !ok {
				return hostbox.Selection{}, hostbox.ErrUnsupported
			}
			return hw.sel, nil
		},
		Pub:        hsPub,
		Ratchet:    func() *release.DigestRatchet { return hsRatchet(t) },
		Proof:      func() *release.ProofGate { return offProofGate(t) },
		OptDir:     f.opt,
		DataDir:    f.data,
		ReceiptDir: f.receipts,
	}
	return f
}

func (f *hrFixture) claims(_ context.Context, host string) (hostclaim.Observed, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.claimReads++
	if host != f.claim.Host {
		return hostclaim.Observed{Host: host, Absent: true}, nil
	}
	raw, err := json.Marshal(f.claim)
	if err != nil {
		return hostclaim.Observed{}, err
	}
	var c hostclaim.Claim
	if err := json.Unmarshal(raw, &c); err != nil {
		return hostclaim.Observed{}, err
	}
	return hostclaim.Observed{Host: host, Claim: &c}, nil
}

func (f *hrFixture) records(_ context.Context, acct, app, host string) (*provision.Record, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if acct != f.rec.Account || app != f.rec.App || host != f.rec.Host {
		return nil, nil
	}
	raw, err := json.Marshal(f.rec)
	if err != nil {
		return nil, err
	}
	var rec provision.Record
	if err := json.Unmarshal(raw, &rec); err != nil {
		return nil, err
	}
	return &rec, nil
}

// journal journals the held render's hash for op, as configured does.
func (f *hrFixture) journal(t *testing.T, op, hash string) {
	t.Helper()
	if err := moves.UpdateJournal(f.data, p1Host, op, func(j *moves.Journal) error {
		j.Mark(landStepRendered, hash)
		return nil
	}); err != nil {
		t.Fatal(err)
	}
}

// request is the agent's request for action, built from the fixture's claim
// and record as the agent builds it from its own reads.
func (f *hrFixture) request(action string) heldReleaseRequest {
	f.mu.Lock()
	defer f.mu.Unlock()
	c := f.claim
	q := heldReleaseRequest{Schema: heldReleaseSchema, RequestID: "req-1", Action: action, Host: p1Host, Account: "bab", App: "app", Version: "v7",
		MoveID: c.Op.ID, HoldOp: c.Op.ID, Box: "uk-box", Region: "uk", Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp,
		Session: c.Holder.Session, Generation: c.Holder.Generation, Kind: c.Op.Kind, Phase: c.Op.Phase}
	if c.Op.Kind == hostclaim.KindRestoring {
		q.HoldOp = c.Op.Intent["rollback"]
	}
	if action == heldReleasePublish {
		q.RenderHash = hrRendered
	}
	return q
}

func (f *hrFixture) set(fn func(f *hrFixture)) {
	f.mu.Lock()
	defer f.mu.Unlock()
	fn(f)
}

// untouched fails unless root had no effect: nothing selected, nothing
// staged, live/<host> absent and no receipt.
func (f *hrFixture) untouched(t *testing.T) {
	t.Helper()
	if n := f.sel.count(); n != 0 {
		t.Fatalf("root selected %d time(s) for a refused request", n)
	}
	if _, err := os.Stat(filepath.Join(f.opt, "versions")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("root staged for a refused request: %v", err)
	}
	if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("root published for a refused request: %v", err)
	}
	if ents, _ := os.ReadDir(f.receipts); len(ents) != 0 {
		t.Fatalf("root wrote a receipt for a refused request: %v", ents)
	}
}

// TestHeldReleaseRootPublishesAtConfigured: a landing's configured step
// asks root to publish the pinned release; root selects held-release for
// the host, publishes live/<host> under the identity naming exactly v7,
// writes the receipt, and the agent's own reading then passes.
func TestHeldReleaseRootPublishesAtConfigured(t *testing.T) {
	ctx := context.Background()
	f := newHRFixture(t)
	resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
	if err != nil {
		t.Fatalf("publish: %v", err)
	}
	if resp.Outcome != heldReleaseOutcomePublished || resp.Replayed || resp.RequestID != "req-1" {
		t.Fatalf("response = %+v", resp)
	}
	if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" {
		t.Fatalf("live = %q, want v7", v)
	}
	if len(f.sel.selected) != 1 || f.sel.selected[0] != p1Host+"/"+movesActionHeldRelease {
		t.Fatalf("selected %v, want one held-release selection", f.sel.selected)
	}
	if len(f.sel.ids) != 1 || f.sel.ids[0] != (hostwrite.Identity{Release: "v7"}) {
		t.Fatalf("published under %v, want the pinned release alone", f.sel.ids)
	}
	if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); err != nil {
		t.Fatalf("no publish receipt: %v", err)
	}
	rt := registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", HoldOp: p1ID}
	lock := func(read func() error) error { return withHeldStageLock(f.data, p1Host, read) }
	if err := release.VerifyHeldLive(ctx, f.s, f.opt, rt, "v7", offProofGate(t), lock); err != nil {
		t.Fatalf("the agent's reading after root's publication: %v", err)
	}
}

// TestHeldReleaseRootPublishesForTheDirtyReturnAndTheProbe: the same path
// serves a landing's probe and a dirty return, at its configured (claimed)
// and its probe, each under its own hold op and render journal.
func TestHeldReleaseRootPublishesForTheDirtyReturnAndTheProbe(t *testing.T) {
	ctx := context.Background()
	asReturn := func(f *hrFixture, phase, sourcePhase string, recovery *provision.MoveInputs) {
		f.claim.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: p1ID, Phase: phase, Exec: "uk-box", Participants: []string{"uk-box", "ie-box"},
			Intent: map[string]string{"source": "uk-box", "target": "ie-box", "to": "ie", "rollback": hrRB}}
		f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveRollingBack, ""
		f.rec.Move.From, f.rec.Move.To = "uk", "ie"
		f.rec.Move.Rollback = &provision.MoveRollback{ID: hrRB, Dirty: true, SourcePhase: sourcePhase, RecoveryInputs: recovery}
		f.row.HoldOp = hrRB
	}
	cases := []struct {
		name string
		set  func(f *hrFixture)
		op   string
	}{
		{"the landing's probe", func(f *hrFixture) {
			f.claim.Op.Phase = provision.LandProbing
			f.rec.Move.LandingPhase = provision.LandProbing
		}, p1ID},
		{"a dirty return's configured", func(f *hrFixture) {
			asReturn(f, hostclaim.PhaseClaimed, provision.RbSourceConfigured, nil)
		}, hrRB},
		{"a dirty return's probe", func(f *hrFixture) {
			asReturn(f, hostclaim.PhaseProbing, provision.RbSourceProbing, &provision.MoveInputs{Release: "v7"})
		}, hrRB},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			f := newHRFixture(t)
			f.set(tc.set)
			f.journal(t, tc.op, hrRendered)
			q := f.request(heldReleasePublish)
			if q.HoldOp != tc.op {
				t.Fatalf("request hold op %q, want %q", q.HoldOp, tc.op)
			}
			resp, err := f.root.Serve(ctx, hrPeer, q)
			if err != nil || resp.Outcome != heldReleaseOutcomePublished {
				t.Fatalf("publish: %+v %v", resp, err)
			}
			if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" {
				t.Fatalf("live = %q, want v7", v)
			}
			if len(f.sel.ids) != 1 || f.sel.ids[0] != (hostwrite.Identity{Release: "v7"}) {
				t.Fatalf("published under %v", f.sel.ids)
			}
		})
	}
}

// TestHeldReleaseRootRefusesBeforeAnyEffect: every request whose tuple is
// not the claim's, whose pins or render are not the record's and the
// journal's, that names another account, app, box or region, or that
// arrives while the claim grants no held write, is refused before root
// selects, stages, publishes or writes a receipt.
func TestHeldReleaseRootRefusesBeforeAnyEffect(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	otherInc := strings.Repeat("c", 64)
	cases := []struct {
		name   string
		action string
		state  func(f *hrFixture)
		q      func(q *heldReleaseRequest)
	}{
		{"a stale incarnation", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Incarnation = otherInc }},
		{"a stale stamp", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Stamp = 6 }},
		{"a stale session", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Session = 2 }},
		{"a stale generation", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Generation = "cur" }},
		{"a stale op", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.MoveID, q.HoldOp = p1ID2, p1ID2 }},
		{"a stale phase", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Phase = provision.LandProbing }},
		{"a stale kind", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Kind = hostclaim.KindReverting }},
		{"changed pins: another version", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Version = "v8" }},
		{"changed pins: the record's", heldReleasePublish, func(f *hrFixture) { f.rec.Move.Inputs.Release = "v8" }, nil},
		{"changed render", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.RenderHash = hrOther }},
		{"a forged account", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Account = "zed" }},
		{"a forged app", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.App = "other" }},
		{"the wrong box", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Box = "ie-box" }},
		{"the wrong region", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Region = "ie" }},
		{"another hold op", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.HoldOp = hrRB }},
		{"the row held by another op", heldReleasePublish, func(f *hrFixture) { f.row.HoldOp = p1ID2 }, nil},
		{"no held row", heldReleasePublish, func(f *hrFixture) { f.row = nil }, nil},
		{"the record's move is another", heldReleasePublish, func(f *hrFixture) { f.rec.Move.ID = p1ID2 }, nil},
		{"the record is not at configured", heldReleasePublish, func(f *hrFixture) { f.rec.Move.LandingPhase = provision.LandLocal }, nil},
		{"this box is not the executor", heldReleasePublish, func(f *hrFixture) { f.claim.Op.Exec = "ie-box" }, nil},
		{"this box is declared lost", heldReleasePublish, func(f *hrFixture) { f.claim.Lost = []string{"uk-box"} }, nil},
		{"a moving claim publishes nothing", heldReleasePublish, func(f *hrFixture) {
			f.claim.Op.Kind, f.claim.Op.Phase = hostclaim.KindMoving, ""
		}, func(q *heldReleaseRequest) { q.Kind, q.Phase = hostclaim.KindMoving, "" }},
		{"a landing at local publishes nothing", heldReleasePublish, func(f *hrFixture) {
			f.claim.Op.Phase, f.rec.Move.LandingPhase = provision.LandLocal, provision.LandLocal
		}, func(q *heldReleaseRequest) { q.Phase = provision.LandLocal }},
		{"no claim", heldReleasePublish, func(f *hrFixture) { f.claim.Host = "other.example.com" }, nil},
		{"stage: the wrong box", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Box = "ie-box" }},
		{"stage: a forged app", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.App = "other" }},
		{"stage: a forged account", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Account = "zed" }},
		{"stage: a stale session", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Session = 9 }},
		{"stage: a stale incarnation", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Incarnation = otherInc }},
		{"stage: changed pins", heldReleaseStage, func(f *hrFixture) { f.rec.Move.Inputs.Release = "v8" }, nil},
		{"stage: another target", heldReleaseStage, func(f *hrFixture) { f.claim.Op.Intent["target"] = "fr-box" }, nil},
		{"stage: a render hash", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.RenderHash = hrRendered }},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			f := newHRFixture(t)
			if tc.state != nil {
				f.set(tc.state)
			}
			q := f.request(tc.action)
			if tc.q != nil {
				tc.q(&q)
			}
			_, err := f.root.Serve(ctx, hrPeer, q)
			if err == nil {
				t.Fatal("root accepted the request")
			}
			if errors.Is(err, errHeldReleaseNotYet) {
				t.Fatalf("refused as not yet: %v", err)
			}
			f.untouched(t)
		})
	}
}

// TestHeldReleaseRootStages: a stage request stages and verifies the pinned
// release into the shared version store and never touches live/<host>, and
// selects nothing: a moving claim grants no held write.
func TestHeldReleaseRootStages(t *testing.T) {
	ctx := context.Background()
	f := newHRFixture(t)
	f.set(func(f *hrFixture) {
		f.claim.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: p1ID, Exec: "ie-box", Participants: []string{"ie-box", "uk-box"},
			Intent: map[string]string{"source": "ie-box", "target": "uk-box", "to": "uk"}}
		f.rec.Move.State, f.rec.Move.LandingPhase, f.rec.Move.Inputs = provision.MoveReady, "", nil
		f.row = nil
	})
	resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleaseStage))
	if err != nil || resp.Outcome != heldReleaseOutcomeStaged {
		t.Fatalf("stage: %+v %v", resp, err)
	}
	if _, err := os.Stat(release.VersionPath(f.opt, "bab", "v7", "app")); err != nil {
		t.Fatalf("v7 not staged: %v", err)
	}
	if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a stage published live/<host>: %v", err)
	}
	if n := f.sel.count(); n != 0 {
		t.Fatalf("a stage selected %d time(s)", n)
	}
	// A publish request against a moving claim is refused: the policy is
	// never widened.
	q := f.request(heldReleasePublish)
	if _, err := f.root.Serve(ctx, hrPeer, q); err == nil {
		t.Fatal("root published under a moving claim")
	}
	if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("live/<host> under a moving claim: %v", err)
	}
}

// TestHeldReleaseRootNotYetWithoutTheRenderJournal (platform-1e Q2): with
// no render-hash journal, or one without the rendered step, root refuses
// with the typed not-yet error at once. It selects nothing and holds no
// lock, so the agent's retry finds the held-stage lock free.
func TestHeldReleaseRootNotYetWithoutTheRenderJournal(t *testing.T) {
	ctx := context.Background()
	for _, tc := range []struct {
		name string
		prep func(t *testing.T, f *hrFixture)
	}{
		{"no journal", func(t *testing.T, f *hrFixture) {
			f.data = t.TempDir()
			f.root.DataDir = f.data
		}},
		{"no rendered step", func(t *testing.T, f *hrFixture) {
			f.data = t.TempDir()
			f.root.DataDir = f.data
			if err := moves.UpdateJournal(f.data, p1Host, p1ID, func(j *moves.Journal) error {
				j.Mark(landStepPreRender, `{"before":"absent","intended":["`+hrRendered+`"]}`)
				return nil
			}); err != nil {
				t.Fatal(err)
			}
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newHRFixture(t)
			tc.prep(t, f)
			start := time.Now()
			_, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
			if !errors.Is(err, errHeldReleaseNotYet) {
				t.Fatalf("publish without the journal: %v, want errHeldReleaseNotYet", err)
			}
			if d := time.Since(start); d > 5*time.Second {
				t.Fatalf("root waited %v for the journal", d)
			}
			f.untouched(t)
			if err := withHeldStageLock(f.data, p1Host, func() error { return nil }); err != nil {
				t.Fatalf("the held-stage lock after a not-yet: %v", err)
			}
		})
	}
}

// TestHeldReleaseRootReplayAndCrashRecovery: a replay after success is a
// no-op that returns the same result (the receipt, keyed by session and
// render hash); a crash between the publication and its receipt is
// recovered by the replay, which publishes the same release again.
func TestHeldReleaseRootReplayAndCrashRecovery(t *testing.T) {
	ctx := context.Background()

	t.Run("a replay after success", func(t *testing.T) {
		f := newHRFixture(t)
		q := f.request(heldReleasePublish)
		first, err := f.root.Serve(ctx, hrPeer, q)
		if err != nil {
			t.Fatal(err)
		}
		reads := f.claimReads
		q.RequestID = "req-2"
		again, err := f.root.Serve(ctx, hrPeer, q)
		if err != nil {
			t.Fatalf("replay: %v", err)
		}
		if again.Outcome != first.Outcome || !again.Replayed {
			t.Fatalf("replay = %+v, first = %+v", again, first)
		}
		if f.sel.count() != 2 || f.claimReads <= reads {
			t.Fatalf("a replay selected %d time(s) and read the claim %d more time(s)", f.sel.count(), f.claimReads-reads)
		}
	})

	t.Run("another session is not a replay", func(t *testing.T) {
		f := newHRFixture(t)
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil {
			t.Fatal(err)
		}
		f.set(func(f *hrFixture) { f.claim.Holder.Session = 4 })
		resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
		if err != nil || resp.Replayed {
			t.Fatalf("a new session's request: %+v %v", resp, err)
		}
		if f.sel.count() != 2 {
			t.Fatalf("selections = %d, want 2", f.sel.count())
		}
	})

	t.Run("a configured receipt never answers the probe", func(t *testing.T) {
		// Fix round 2, P3-1: the receipt binds the action, the kind and the
		// phase, so a landing's configured publication does not answer its
		// probing request within the same holder tuple: the probe's request
		// runs the whole path.
		f := newHRFixture(t)
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil {
			t.Fatal(err)
		}
		f.set(func(f *hrFixture) {
			f.claim.Op.Phase = provision.LandProbing
			f.rec.Move.LandingPhase = provision.LandProbing
		})
		reads := f.claimReads
		resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
		if err != nil || resp.Replayed {
			t.Fatalf("the probe's request: %+v %v, want the whole path", resp, err)
		}
		if f.sel.count() != 2 || f.claimReads == reads {
			t.Fatalf("selections %d, claim reads +%d: the probe was answered from configured's receipt", f.sel.count(), f.claimReads-reads)
		}
	})

	t.Run("a stale request is not answered from an old receipt", func(t *testing.T) {
		f := newHRFixture(t)
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil {
			t.Fatal(err)
		}
		q := f.request(heldReleasePublish)
		q.Version = "v8"
		if _, err := f.root.Serve(ctx, hrPeer, q); err == nil {
			t.Fatal("another version was answered from the receipt")
		}
	})

	t.Run("a receipt that does not persist is no success", func(t *testing.T) {
		// Fix round 1, item 1: root answers published only once the receipt
		// that answers a replay has persisted. Without it the answer is an
		// error; the next identical request runs the whole path again (the
		// same render, the same selection checks), publishing the same
		// release, and still errs until a receipt persists. Then a replay
		// is answered from the receipt with no selection.
		f := newHRFixture(t)
		blocker := filepath.Join(t.TempDir(), "not-a-dir")
		if err := os.WriteFile(blocker, nil, 0o600); err != nil {
			t.Fatal(err)
		}
		good := f.root.ReceiptDir
		f.root.ReceiptDir = filepath.Join(blocker, "held-release")
		q := f.request(heldReleasePublish)
		for i := 1; i <= 2; i++ {
			resp, err := f.root.Serve(ctx, hrPeer, q)
			if err == nil || resp.Outcome != "" {
				t.Fatalf("attempt %d with no receipt: %+v %v, want an error", i, resp, err)
			}
			if !errors.Is(err, errHeldReleaseReceipt) {
				t.Fatalf("attempt %d: %v, want errHeldReleaseReceipt", i, err)
			}
			if f.sel.count() != i || len(f.sel.ids) != i || f.sel.ids[i-1] != (hostwrite.Identity{Release: "v7"}) {
				t.Fatalf("attempt %d: selections %d under %v, want each attempt through the whole path", i, f.sel.count(), f.sel.ids)
			}
		}
		// A retry is held to the same render: a changed journal refuses it
		// before any selection.
		f.journal(t, p1ID, hrOther)
		if _, err := f.root.Serve(ctx, hrPeer, q); err == nil || f.sel.count() != 2 {
			t.Fatalf("a retry under another render: %v, selections %d", err, f.sel.count())
		}
		f.journal(t, p1ID, hrRendered)
		f.root.ReceiptDir = good
		resp, err := f.root.Serve(ctx, hrPeer, q)
		if err != nil || resp.Outcome != heldReleaseOutcomePublished || resp.Replayed {
			t.Fatalf("once the receipt persists: %+v %v", resp, err)
		}
		again, err := f.root.Serve(ctx, hrPeer, q)
		if err != nil || !again.Replayed || f.sel.count() != 4 {
			t.Fatalf("the replay: %+v %v, selections %d", again, err, f.sel.count())
		}
		if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" {
			t.Fatalf("live = %q", v)
		}
	})

	t.Run("a crash between the publication and the receipt", func(t *testing.T) {
		f := newHRFixture(t)
		crashed := false
		f.root.afterPublish = func(string) error {
			if !crashed {
				crashed = true
				return errCrash
			}
			return nil
		}
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, errCrash) {
			t.Fatalf("first publish: %v, want the crash", err)
		}
		if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" {
			t.Fatalf("live = %q after the crash", v)
		}
		if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("a receipt before the crash point: %v", err)
		}
		resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
		if err != nil || resp.Outcome != heldReleaseOutcomePublished || resp.Replayed {
			t.Fatalf("recovery: %+v %v", resp, err)
		}
		if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" || f.sel.count() != 2 {
			t.Fatalf("after recovery live = %q, selections %d", v, f.sel.count())
		}
		if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); err != nil {
			t.Fatalf("no receipt after recovery: %v", err)
		}
	})
}

// TestHeldReleaseRequestShape: root refuses a request whose shape it cannot
// trust before any read: the schema, the action, the names, the ids, the
// incarnation, the stamp and the render hash.
func TestHeldReleaseRequestShape(t *testing.T) {
	f := newHRFixture(t)
	for name, mutate := range map[string]func(q *heldReleaseRequest){
		"schema":           func(q *heldReleaseRequest) { q.Schema = "carlos.moves.held-stage/1" },
		"action":           func(q *heldReleaseRequest) { q.Action = "install" },
		"no request id":    func(q *heldReleaseRequest) { q.RequestID = "" },
		"host":             func(q *heldReleaseRequest) { q.Host = "../etc" },
		"account":          func(q *heldReleaseRequest) { q.Account = "Bab/x" },
		"app":              func(q *heldReleaseRequest) { q.App = "a/../b" },
		"version":          func(q *heldReleaseRequest) { q.Version = "../v7" },
		"move id":          func(q *heldReleaseRequest) { q.MoveID = "mv_x" },
		"hold op":          func(q *heldReleaseRequest) { q.HoldOp = "../" + p1ID },
		"box":              func(q *heldReleaseRequest) { q.Box = "" },
		"region":           func(q *heldReleaseRequest) { q.Region = "UK!" },
		"incarnation":      func(q *heldReleaseRequest) { q.Incarnation = "x" },
		"stamp":            func(q *heldReleaseRequest) { q.Stamp = 0 },
		"kind":             func(q *heldReleaseRequest) { q.Kind = hostclaim.KindServing },
		"render hash":      func(q *heldReleaseRequest) { q.RenderHash = "sha256:x" },
		"no render hash":   func(q *heldReleaseRequest) { q.RenderHash = "" },
		"oversized phase":  func(q *heldReleaseRequest) { q.Phase = strings.Repeat("p", 200) },
		"oversized gen id": func(q *heldReleaseRequest) { q.Generation = strings.Repeat("g", 300) },
	} {
		t.Run(name, func(t *testing.T) {
			q := f.request(heldReleasePublish)
			mutate(&q)
			reads := f.claimReads
			if _, err := f.root.Serve(context.Background(), hrPeer, q); !errors.Is(err, errHeldReleaseShape) {
				t.Fatalf("serve: %v, want errHeldReleaseShape", err)
			}
			if f.claimReads != reads {
				t.Fatal("a malformed request reached the claim")
			}
		})
	}
	t.Run("root refuses a peer with no identity", func(t *testing.T) {
		if _, err := f.root.Serve(context.Background(), StopPeer{}, f.request(heldReleasePublish)); !errors.Is(err, ErrClaimStopIdentity) {
			t.Fatalf("serve: %v", err)
		}
		f.untouched(t)
	})
	t.Run("the body is bounded and strict", func(t *testing.T) {
		raw, _ := json.Marshal(f.request(heldReleasePublish))
		if _, err := decodeHeldReleaseRequest(strings.NewReader(strings.TrimSuffix(string(raw), "}") + `,"extra":1}`)); !errors.Is(err, errHeldReleaseShape) {
			t.Fatalf("unknown field: %v", err)
		}
		if _, err := decodeHeldReleaseRequest(strings.NewReader(string(raw) + `{}`)); !errors.Is(err, errHeldReleaseShape) {
			t.Fatalf("trailing data: %v", err)
		}
		if _, err := decodeHeldReleaseRequest(strings.NewReader(strings.Repeat(" ", heldReleaseRequestMax+1) + string(raw))); !errors.Is(err, errHeldReleaseShape) {
			t.Fatalf("oversized: %v", err)
		}
		if q, err := decodeHeldReleaseRequest(strings.NewReader(string(raw))); err != nil || q.Host != p1Host {
			t.Fatalf("a good body: %+v %v", q, err)
		}
	})
}

// TestHeldReleaseRootSelectionMustBeTheRequests: the claim moving on between
// root's read and its own selection (a new session admitted) is refused
// before anything is published: the selection's tuple must be the
// request's. A selection the claim refuses publishes nothing either.
func TestHeldReleaseRootSelectionMustBeTheRequests(t *testing.T) {
	ctx := context.Background()
	t.Run("the claim moved on", func(t *testing.T) {
		f := newHRFixture(t)
		works := f.root.Works
		f.root.Works = func(ctx context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error) {
			f.set(func(f *hrFixture) { f.claim.Holder.Session = 4 })
			return works(ctx, actor, host, action)
		}
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("publish across a session change: %v, want hostbox.ErrChanged", err)
		}
		if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("published across a session change: %v", err)
		}
		if len(f.sel.ids) != 0 {
			t.Fatalf("wrote under %v", f.sel.ids)
		}
	})
	t.Run("the selection is refused", func(t *testing.T) {
		f := newHRFixture(t)
		f.sel.refuse = hostbox.ErrChanged
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("publish with the selection refused: %v", err)
		}
		if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("published with the selection refused: %v", err)
		}
	})
}

// TestHeldReleaseRouteIsMovesOnly: with CARLOS_MOVES_PHASE1 off the claim
// root's handler is returned unchanged, so the route refuses as on main.
func TestHeldReleaseRouteIsMovesOnly(t *testing.T) {
	h := ClaimRootWithLaunch{}
	got, err := WithProductionMovesHeldRelease(context.Background(), h, nil, func(string) string { return "" })
	if err != nil {
		t.Fatal(err)
	}
	if r, ok := got.(ClaimRootWithLaunch); !ok || r.HeldRelease != nil {
		t.Fatalf("moves off wired %#v", got)
	}
}

// TestHeldReleaseRootRefusesUnjournalledRecoveryProbes (fix round 2, P2-1,
// platform-1e's authority review): a revert's probe and a clean return's
// probe journal no render under any rule root checks, so their absent
// journal cannot be what refuses them: the journal is agent-written. Root
// refuses both rows outright, even with a planted journal and a matching
// request, and publishes nothing. A dirty return's probe, whose configured
// step journalled its render, still publishes.
func TestHeldReleaseRootRefusesUnjournalledRecoveryProbes(t *testing.T) {
	ctx := context.Background()
	recovery := func(f *hrFixture) {
		f.claim.Op.Exec, f.claim.Op.Participants = "uk-box", []string{"uk-box", "ie-box"}
		f.claim.Op.Intent = map[string]string{"source": "uk-box", "target": "ie-box", "to": "ie"}
		f.claim.Op.Phase = hostclaim.PhaseProbing
		f.rec.Move.From, f.rec.Move.To, f.rec.Move.LandingPhase = "uk", "ie", ""
	}
	for _, tc := range []struct {
		name string
		set  func(f *hrFixture)
		hold string
	}{
		{"a revert's probe", func(f *hrFixture) {
			recovery(f)
			f.claim.Op.Kind = hostclaim.KindReverting
			f.claim.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing}
			f.rec.Move.State = provision.MoveReverting
			f.rec.Move.RecoveryInputs = &provision.MoveInputs{Release: "v7"}
		}, p1ID},
		{"a clean return's probe", func(f *hrFixture) {
			recovery(f)
			f.claim.Op.Kind = hostclaim.KindRestoring
			f.claim.Op.Intent["rollback"] = hrRB
			f.rec.Move.State = provision.MoveRollingBack
			f.rec.Move.Rollback = &provision.MoveRollback{ID: hrRB, SourcePhase: provision.RbSourceProbing, RecoveryInputs: &provision.MoveInputs{Release: "v7"}}
			f.row.HoldOp = hrRB
		}, hrRB},
	} {
		t.Run(tc.name+" with a planted journal", func(t *testing.T) {
			f := newHRFixture(t)
			f.set(tc.set)
			// A compromised agent plants the render it then sends.
			f.journal(t, tc.hold, hrRendered)
			q := f.request(heldReleasePublish)
			if q.HoldOp != tc.hold {
				t.Fatalf("request hold op %q, want %q", q.HoldOp, tc.hold)
			}
			_, err := f.root.Serve(ctx, hrPeer, q)
			if !errors.Is(err, release.ErrStaleHeldStage) {
				t.Fatalf("publish: %v, want a refusal", err)
			}
			f.untouched(t)
		})
	}
}

// TestHeldReleaseRootBoundsConcurrency (fix round 2, P3-4): root serves
// one request per host at a time, and at most heldReleaseConcurrency in
// all, so the agent cannot pile up downloads inside instance-control. A
// request over either bound is refused at once as busy (503, which the
// agent's client reads as unavailable and retries), before any read; a
// slot is free again once its request returns.
func TestHeldReleaseRootBoundsConcurrency(t *testing.T) {
	ctx := context.Background()
	hold := func(f *hrFixture) (entered chan string, release func()) {
		entered = make(chan string, 16)
		gate := make(chan struct{})
		claims := f.root.Claims
		f.root.Claims = func(ctx context.Context, host string) (hostclaim.Observed, error) {
			entered <- host
			<-gate
			return claims(ctx, host)
		}
		var once sync.Once
		release = func() { once.Do(func() { close(gate) }) }
		t.Cleanup(release)
		return entered, release
	}
	busy := func(t *testing.T, err error) {
		t.Helper()
		if !errors.Is(err, errHeldReleaseBusy) {
			t.Fatalf("over the bound: %v, want errHeldReleaseBusy", err)
		}
		if got := heldReleaseStatus(err); got != 503 {
			t.Fatalf("busy answers %d, want 503", got)
		}
	}

	t.Run("one per host", func(t *testing.T) {
		f := newHRFixture(t)
		entered, release := hold(f)
		done := make(chan error, 1)
		go func() {
			_, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish))
			done <- err
		}()
		<-entered
		reads := f.claimReads
		_, err := f.root.Serve(ctx, hrPeer, f.request(heldReleaseStage))
		busy(t, err)
		if f.claimReads != reads || len(entered) != 0 {
			t.Fatal("a busy request read the claim")
		}
		release()
		if err := <-done; err != nil {
			t.Fatalf("the first request: %v", err)
		}
		if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil {
			t.Fatalf("after the first returned: %v", err)
		}
	})

	t.Run("a bound across hosts", func(t *testing.T) {
		f := newHRFixture(t)
		entered, release := hold(f)
		other := func(i int) heldReleaseRequest {
			q := f.request(heldReleaseStage)
			q.Host = fmt.Sprintf("h%d.go.example.com", i)
			return q
		}
		done := make(chan error, heldReleaseConcurrency)
		for i := 0; i < heldReleaseConcurrency; i++ {
			go func(i int) {
				_, err := f.root.Serve(ctx, hrPeer, other(i))
				done <- err
			}(i)
			<-entered
		}
		_, err := f.root.Serve(ctx, hrPeer, other(99))
		busy(t, err)
		release()
		for i := 0; i < heldReleaseConcurrency; i++ {
			if err := <-done; errors.Is(err, errHeldReleaseBusy) {
				t.Fatalf("a held request: %v", err)
			}
		}
		if _, err := f.root.Serve(ctx, hrPeer, other(99)); errors.Is(err, errHeldReleaseBusy) {
			t.Fatalf("after the slots freed: %v", err)
		}
	})
}