package host

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"slices"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/cgroup"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/layout"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The offline operator move's box side (Task 3 of
// docs/superpowers/plans/2026-10-20-offline-operator-move.md), tested
// through the functions the agent calls: newMovesWorker, armOffline,
// reconcileSerializedWith, the production edges-file writer and movesPass,
// over one DirStore, real registries and real activator services (the
// package's fake wake backend, which only the fence needs).

const (
	offHost = "shop.example.com"
	offID   = "mv_zyxwvutsrqponmlkjihgfedcba"
	offID2  = "mv_abcdefghijklmnopqrstuvwxy2"
)

type offWorld struct {
	t    *testing.T
	s    release.Store
	acct string
	mu   sync.Mutex
	now  time.Time
}

func (wd *offWorld) clock() time.Time {
	wd.mu.Lock()
	defer wd.mu.Unlock()
	return wd.now
}

func (wd *offWorld) advance(d time.Duration) {
	wd.mu.Lock()
	wd.now = wd.now.Add(d)
	wd.mu.Unlock()
}

// newOffWorld is a deployment with one account, one app and one instance
// record for offHost at the home placement ("").
func newOffWorld(t *testing.T) *offWorld {
	t.Helper()
	t.Setenv("CARLOS_HOST_CLAIM", "") // the appconfig and provision guards read the process env
	ctx := context.Background()
	wd := &offWorld{t: t, s: release.NewDirStore(t.TempDir()), acct: domains.Sqid(2), now: time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)}
	manifest, err := json.Marshal(map[string]any{"id": 2, "sqid": wd.acct, "name": "fixture", "created_at": wd.now})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := wd.s.Put(ctx, "accounts/"+wd.acct+"/account.json", manifest, ""); err != nil {
		t.Fatal(err)
	}
	if err := provision.PutManifest(ctx, wd.s, wd.acct, "app", provision.Manifest{Schema: 1, App: "app", Domain: "example.com"}); err != nil {
		t.Fatal(err)
	}
	wd.create(offHost)
	return wd
}

func (wd *offWorld) create(host string) {
	wd.t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0], Account: wd.acct, App: "app", Region: "", Channel: "stable", CreatedAt: wd.now}
	if err := provision.CreateRecord(context.Background(), wd.s, rec); err != nil {
		wd.t.Fatal(err)
	}
}

// move applies one offline transition to offHost's record.
func (wd *offWorld) move(mut func(*provision.Record) error) provision.Record {
	wd.t.Helper()
	return wd.moveHost(offHost, mut)
}

func (wd *offWorld) moveHost(host string, mut func(*provision.Record) error) provision.Record {
	wd.t.Helper()
	rec, err := provision.SetRecordMove(context.Background(), wd.s, wd.acct, "app", host, mut)
	if err != nil {
		wd.t.Fatal(err)
	}
	return *rec
}

func (wd *offWorld) record() provision.Record {
	wd.t.Helper()
	rec, err := provision.GetRecord(context.Background(), wd.s, wd.acct, "app", offHost)
	if err != nil {
		wd.t.Fatal(err)
	}
	return *rec
}

// The legal path to a state, from the record at rest.
func (wd *offWorld) start() {
	wd.move(provision.OfflineStart(offID, "", "uk", "op@example.com", wd.clock()))
}
func (wd *offWorld) seal() { wd.move(provision.OfflineSeal(offID, "ie-a", []string{"ie-a", "ie-b"})) }
func (wd *offWorld) sealed() {
	wd.move(provision.OfflineSealedBy(offID, "ie-a", provision.MoveSource{}))
}
func (wd *offWorld) flip()   { wd.move(provision.OfflineFlip(offID)) }
func (wd *offWorld) landed() { wd.move(provision.OfflineLandedBy(offID, "uk")) }
func (wd *offWorld) open()   { wd.move(provision.OfflineOpenTransition(offID, wd.clock())) }
func (wd *offWorld) finishing() {
	wd.move(provision.OfflineFinishingTransition(offID))
}
func (wd *offWorld) abort() { wd.move(provision.OfflineAbort(offID)) }

// report reads one box's report for offID, or fails.
func (wd *offWorld) report(role, box string) moves.OfflineReport {
	wd.t.Helper()
	r, ok := wd.reportIf(role, box)
	if !ok {
		wd.t.Fatalf("no %s report from %s", role, box)
	}
	return r
}

func (wd *offWorld) reportIf(role, box string) (moves.OfflineReport, bool) {
	wd.t.Helper()
	body, _, err := wd.s.Get(context.Background(), moves.OfflineReportKey(offHost, offID, role, box))
	if errors.Is(err, os.ErrNotExist) {
		return moves.OfflineReport{}, false
	}
	if err != nil {
		wd.t.Fatal(err)
	}
	var r moves.OfflineReport
	if err := json.Unmarshal(body, &r); err != nil {
		wd.t.Fatal(err)
	}
	if err := moves.ValidateOfflineReport(r); err != nil {
		wd.t.Fatal(err)
	}
	return r, true
}

// offBox is one box: its own data directory, registry, activator and moves
// worker, armed as armMovesPass arms it (armOffline).
type offBox struct {
	wd           *offWorld
	name, region string
	dd, opt, cg  string
	env          map[string]string
	reg          *registry.Registry
	svc          *activator.Service
	w            *movesWorker
	nested       bool
	last         []byte
}

func (wd *offWorld) box(name, region string, env map[string]string) *offBox {
	wd.t.Helper()
	b := &offBox{wd: wd, name: name, region: region, dd: wd.t.TempDir(), opt: wd.t.TempDir(), cg: wd.t.TempDir(), env: env}
	reg, err := registry.Open(filepath.Join(b.dd, "registry.db"))
	if err != nil {
		wd.t.Fatal(err)
	}
	wd.t.Cleanup(func() { reg.Close() })
	b.reg = reg
	b.restart()
	return b
}

func (b *offBox) getenv(k string) string {
	if v, ok := b.env[k]; ok {
		return v
	}
	if k == "CARLOS_HIBERNATE_BUCKET" {
		return "data"
	}
	return ""
}

// restart is an agent restart: a fresh activator (no fences) and a fresh
// worker over the same registry and data directory.
func (b *offBox) restart() {
	b.svc = activator.New(activator.Config{Bucket: "data", Idle: time.Hour}, &fakeWakeBackend{}, b.reg)
	cfg := moveCfg{Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock}
	armOffline(&cfg, b.dd, b.reg, b.svc, nil, hostwrite.Config{}, b.getenv)
	// The stage-requested marker is root's business, and its throttled
	// timer would outlive the test's directories.
	cfg.Offline = newOfflineView(cfg, nil)
	cfg.ReleaseHold = releaseHeld(b.reg, nil)
	cfg.Procs = func(host string) (int, error) { return cgroup.HostProcs(b.cg, host, "") }
	b.w = newMovesWorker(cfg)
}

func (b *offBox) reconcile() {
	b.wd.t.Helper()
	cfg := provision.ReconcileConfig{Region: b.region, DataDir: b.dd}
	if b.nested {
		cfg.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }
	} else {
		cfg.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeFlat, nil }
	}
	if err := reconcileSerializedWith(context.Background(), b.wd.s, b.reg, cfg, b.w); err != nil {
		b.wd.t.Fatal(err)
	}
}

func (b *offBox) edges() {
	b.wd.t.Helper()
	if err := edgesFileWrite(context.Background(), b.wd.s, b.reg, edgesFile{Box: b.name, Placement: b.region}, "", &b.last, false); err != nil {
		b.wd.t.Fatal(err)
	}
}

func (b *offBox) pass() {
	b.wd.t.Helper()
	if err := movesPass(context.Background(), b.wd.s, b.w); err != nil {
		b.wd.t.Fatal(err)
	}
}

// tick is the agent's order on a serial bump: reconcile, the edges file,
// the moves pass.
func (b *offBox) tick() {
	b.wd.t.Helper()
	b.reconcile()
	b.edges()
	b.pass()
}

func (b *offBox) row(host string) *registry.Route {
	b.wd.t.Helper()
	rt, err := b.reg.Get(host)
	if err != nil {
		b.wd.t.Fatal(err)
	}
	return rt
}

// lists reports whether this box's published edges file lists host.
func (b *offBox) lists(host string) bool {
	b.wd.t.Helper()
	body, _, err := b.wd.s.Get(context.Background(), edgesFileKey(b.name))
	if err != nil {
		b.wd.t.Fatal(err)
	}
	var f edgesFile
	if err := json.Unmarshal(body, &f); err != nil {
		b.wd.t.Fatal(err)
	}
	_, ok := f.Hosts[host]
	return ok
}

func writeFile(t *testing.T, p, body string) {
	t.Helper()
	if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(p, []byte(body), 0o644); err != nil {
		t.Fatal(err)
	}
}

// --- the hold ---------------------------------------------------------------

// offlineHold's table: a pure function of the record and the box's
// placement. The home's "" and a named region; every state.
func TestOfflineHoldIsAFunctionOfTheRecord(t *testing.T) {
	home := moveCfg{Region: ""}
	uk := moveCfg{Region: "uk"}
	fr := moveCfg{Region: "fr"}
	rec := func(region, state string) provision.Record {
		return provision.Record{Host: offHost, Region: region, Move: &provision.MoveState{Kind: provision.MoveKindOffline, ID: offID, From: "", To: "uk", State: state}}
	}
	cases := []struct {
		name               string
		box                moveCfg
		rec                provision.Record
		wantStatus, wantOp string
	}{
		{"source stopping", home, rec("", provision.OfflineStopping), "moving", offID},
		{"source sealing", home, rec("", provision.OfflineSealing), "moving", offID},
		{"source sealed", home, rec("", provision.OfflineSealed), "moving", offID},
		{"source reverting", home, rec("", provision.OfflineReverting), "", ""},
		{"source after the flip", home, rec("uk", provision.OfflineLanding), "", ""},
		{"target before the flip", uk, rec("", provision.OfflineSealed), "", ""},
		{"target landing", uk, rec("uk", provision.OfflineLanding), "landing", offID},
		{"target landed", uk, rec("uk", provision.OfflineLanded), "landing", offID},
		{"target landing, region not yet the target's", uk, rec("", provision.OfflineLanding), "", ""},
		{"target open", uk, rec("uk", provision.OfflineOpen), "", ""},
		{"target finishing", uk, rec("uk", provision.OfflineFinishing), "", ""},
		{"target reverting", uk, rec("", provision.OfflineReverting), "", ""},
		{"a third region", fr, rec("", provision.OfflineStopping), "", ""},
		{"a third region after the flip", fr, rec("uk", provision.OfflineLanding), "", ""},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			status, op, ok := offlineHold(c.box)(c.rec)
			if status != c.wantStatus || op != c.wantOp || ok != (c.wantStatus != "") {
				t.Fatalf("hold = (%q, %q, %v), want (%q, %q)", status, op, ok, c.wantStatus, c.wantOp)
			}
		})
	}
	// A record at rest, a legacy move and a Phase 1 move are never held.
	for _, m := range []*provision.MoveState{nil, {To: "uk", State: provision.MoveCutting}, {ID: offID, From: "", To: "uk", State: provision.OfflineStopping}} {
		if _, _, ok := offlineHold(home)(provision.Record{Host: offHost, Move: m}); ok {
			t.Fatalf("held: %+v", m)
		}
	}
}

// B1, the pinning test: a fresh target worker whose keep view has never
// seen the move (built from a list with no offline record in it, as a
// box's view is before the flip) mints the row held "landing" on its first
// reconcile after the flip. And with the view never built at all, reconcile
// through the offline hooks alone does the same.
func TestOfflineFreshTargetMintsTheLandingRowHeld(t *testing.T) {
	wd := newOffWorld(t)
	wd.start()
	wd.seal()
	wd.sealed()
	wd.flip()

	uk := wd.box("uk-box", "uk", nil)
	uk.w.cfg.Offline.refresh(nil, nil) // stale: no offline move in it
	uk.reconcile()
	rt := uk.row(offHost)
	if rt == nil || rt.Status != "landing" || rt.HoldOp != offID {
		t.Fatalf("after the flip the target minted %+v, want held landing by %s", rt, offID)
	}
	uk.edges()
	if uk.lists(offHost) {
		t.Fatal("the target's edges file lists a held landing row")
	}

	// The hooks with a view that was never built, straight into
	// provision.Reconcile: the hold needs no view.
	fresh := wd.box("uk-box-2", "uk", nil)
	if fresh.w.cfg.Offline.built() {
		t.Fatal("fixture: the view is built")
	}
	cfg := withOfflineHolds(provision.ReconcileConfig{Region: "uk", DataDir: fresh.dd}, offlineHold(fresh.w.cfg), fresh.w.cfg.Offline)
	if err := provision.Reconcile(context.Background(), wd.s, fresh.reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt := fresh.row(offHost); rt == nil || rt.Status != "landing" || rt.HoldOp != offID {
		t.Fatalf("an unbuilt view: minted %+v", rt)
	}
}

// --- the source boxes -------------------------------------------------------

// start holds the row on both source boxes (one minting it held, one
// holding the row it has, keeping its other columns), both report held,
// fenced and quiesced, and neither edges file lists the host. An agent
// restart re-fences and re-reports. After the flip the rows stay, kept.
func TestOfflineStartHoldsEverySourceRow(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	b := wd.box("ie-b", "", nil)
	uk := wd.box("uk-box", "uk", nil)
	a.tick() // ie-a serves the host before the move; ie-b has no row yet
	if rt := a.row(offHost); rt == nil || rt.Status != "ready" || !a.lists(offHost) {
		t.Fatalf("fixture: ie-a does not serve the host: %+v", rt)
	}
	if err := a.reg.SetReplication(offHost, "daily"); err != nil {
		t.Fatal(err)
	}
	if err := a.reg.SetNextDue(offHost, 1234); err != nil {
		t.Fatal(err)
	}
	before := *a.row(offHost)

	wd.start()
	a.tick()
	b.tick()
	uk.tick()
	for _, bx := range []*offBox{a, b} {
		rt := bx.row(offHost)
		if rt == nil || rt.Status != "moving" || rt.HoldOp != offID {
			t.Fatalf("%s: row %+v, want held moving by the move", bx.name, rt)
		}
		if bx.lists(offHost) {
			t.Fatalf("%s's edges file lists a held host", bx.name)
		}
		if !bx.svc.Fenced(offHost) {
			t.Fatalf("%s: no wake fence", bx.name)
		}
		r := wd.report(moves.OfflineRoleSource, bx.name)
		if !r.Held || !r.Fenced || !r.Quiesced || !r.RowPresent || r.Procs != 0 || r.LayoutJournal || r.Phase != moves.OfflinePhaseStopped || !r.At.Equal(wd.clock()) {
			t.Fatalf("%s: report %+v", bx.name, r)
		}
	}
	after := *a.row(offHost)
	after.Status, after.HoldOp = before.Status, before.HoldOp
	if !reflect.DeepEqual(after, before) {
		t.Fatalf("holding the existing row changed its other columns:\n before %+v\n after  %+v", before, after)
	}
	if uk.row(offHost) != nil {
		t.Fatal("the target minted a row before the flip")
	}
	if _, ok := wd.reportIf(moves.OfflineRoleSource, "uk-box"); ok {
		t.Fatal("the target wrote a source report")
	}

	// An agent restart: the fence is gone with the process, and the next
	// pass installs it again and rewrites the report.
	a.restart()
	if a.svc.Fenced(offHost) {
		t.Fatal("fixture: a fresh activator is fenced")
	}
	wd.advance(time.Minute)
	a.tick()
	if !a.svc.Fenced(offHost) {
		t.Fatal("after a restart the pass did not fence again")
	}
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); !r.Fenced || !r.At.Equal(wd.clock()) {
		t.Fatalf("after a restart: %+v", r)
	}

	// The flip: the record names uk; both source rows stay, held, and keep
	// reporting; the target mints its row held landing.
	wd.seal()
	wd.sealed()
	wd.flip()
	wd.advance(time.Minute)
	for _, bx := range []*offBox{a, b, uk} {
		bx.tick()
	}
	for _, bx := range []*offBox{a, b} {
		if rt := bx.row(offHost); rt == nil || rt.Status != "moving" || rt.HoldOp != offID {
			t.Fatalf("%s after the flip: %+v, want the row kept, held", bx.name, rt)
		}
		if r := wd.report(moves.OfflineRoleSource, bx.name); !r.Held || !r.Fenced || !r.At.Equal(wd.clock()) {
			t.Fatalf("%s after the flip: %+v", bx.name, r)
		}
	}
	if rt := uk.row(offHost); rt == nil || rt.Status != "landing" || rt.HoldOp != offID {
		t.Fatalf("target after the flip: %+v", rt)
	}
	if uk.lists(offHost) {
		t.Fatal("the target lists the landing row")
	}
}

// What the report reads from disk: a box with only a meta dir has local
// state; a process left in the host's cgroup and a layout journal show.
func TestOfflineSourceReportReadsTheBox(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	b := wd.box("ie-b", "", nil)
	a.tick()
	wd.start()
	// ie-b has no DB file, only the activator's meta dir with one LTX file.
	meta := activatorMetaDir(provision.DefaultDBPath(b.dd, offHost))
	writeFile(t, filepath.Join(meta, "ltx", "0", "0000000000000001-0000000000000007.ltx"), "x")
	// ie-a: files, env, a process in the host's exec cgroup, a journal.
	writeFile(t, filepath.Join(a.dd, offHost, "instance.key"), "k")
	writeFile(t, filepath.Join(a.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=k\n")
	writeFile(t, filepath.Join(a.cg, cgroup.ExecName(offHost), "cgroup.procs"), "4242\n")
	writeFile(t, provision.LayoutJournalPath(a.dd, offHost), "{}")
	a.tick()
	b.tick()
	rb := wd.report(moves.OfflineRoleSource, "ie-b")
	if !rb.LocalDB || rb.LocalTXID != 7 || rb.Files || rb.Env || rb.Sidecar {
		t.Fatalf("a meta dir alone: %+v", rb)
	}
	ra := wd.report(moves.OfflineRoleSource, "ie-a")
	if ra.LocalDB || !ra.Files || !ra.Env || ra.Procs != 1 || !ra.LayoutJournal {
		t.Fatalf("ie-a: %+v", ra)
	}
	// A box that cannot count its processes reports -1, never 0.
	a.w.cfg.Procs = func(string) (int, error) { return 0, cgroup.ErrUnknown }
	a.pass()
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); r.Procs != -1 {
		t.Fatalf("unknown processes reported as %d", r.Procs)
	}
	// A sidecar file shows.
	writeFile(t, filepath.Join(b.dd, offHost+"-sidecar.env"), "X=1\n")
	b.pass()
	if r := wd.report(moves.OfflineRoleSource, "ie-b"); !r.Sidecar {
		t.Fatalf("sidecar not seen: %+v", r)
	}
}

// offline-reverting, before and after the flip: every source row is
// released and served again, the fence is lifted, the boxes report
// reverted, and the next reconcile does not hold the rows again.
func TestOfflineRevertReleasesEverySourceRow(t *testing.T) {
	for _, afterFlip := range []bool{false, true} {
		t.Run(map[bool]string{false: "before the flip", true: "after the flip"}[afterFlip], func(t *testing.T) {
			wd := newOffWorld(t)
			a := wd.box("ie-a", "", nil)
			b := wd.box("ie-b", "", nil)
			uk := wd.box("uk-box", "uk", nil)
			a.tick()
			b.tick()
			wd.start()
			if afterFlip {
				wd.seal()
				wd.sealed()
				wd.flip()
			}
			for _, bx := range []*offBox{a, b, uk} {
				bx.tick()
			}
			wd.abort()
			for i := 0; i < 2; i++ {
				for _, bx := range []*offBox{a, b, uk} {
					bx.tick()
				}
			}
			for _, bx := range []*offBox{a, b} {
				rt := bx.row(offHost)
				if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
					t.Fatalf("%s: %+v, want ready and unheld", bx.name, rt)
				}
				if !bx.lists(offHost) {
					t.Fatalf("%s does not serve the host again", bx.name)
				}
				if bx.svc.Fenced(offHost) {
					t.Fatalf("%s: the fence is still up", bx.name)
				}
				if r := wd.report(moves.OfflineRoleSource, bx.name); r.Phase != moves.OfflinePhaseReverted || r.Held {
					t.Fatalf("%s: %+v", bx.name, r)
				}
			}
			// The move finishes; nothing holds the rows again.
			wd.move(provision.OfflineFinish(offID, wd.clock()))
			for _, bx := range []*offBox{a, b} {
				bx.tick()
				if rt := bx.row(offHost); rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
					t.Fatalf("%s after finish: %+v", bx.name, rt)
				}
			}
		})
	}
}

// offline-open on a flat source: every path of the source-retire row that
// is there is renamed into the park directory byte for byte, nothing is
// deleted, the fence is lifted, and the parked report lists every path with
// its action. A process left in the host's cgroup stops the park first.
// offline-finishing then removes the row and the box reports gone.
func TestOfflineOpenParksTheFlatSource(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	uk := wd.box("uk-box", "uk", nil)
	a.tick()
	db := a.row(offHost).DBPath
	contents := map[string]string{
		db:          "sqlite",
		db + "-wal": "wal",
		filepath.Join(activatorMetaDir(db), "ltx", "0", "0000000000000001-0000000000000003.ltx"): "ltx",
		filepath.Join(a.dd, offHost, "instance.key"):                                             "key",
		filepath.Join(a.dd, offHost+".env"):                                                      "CARLOS_INSTANCE_KEY=k\n",
		filepath.Join(a.dd, offHost+".config.env"):                                               "SECRET=s\n",
		filepath.Join(a.dd, "programs", offHost+".json"):                                         `{"programs":[]}`,
		filepath.Join(a.opt, "live", offHost, "app"):                                             "binary",
		filepath.Join(a.dd, "neighbour.example.com.db"):                                          "not ours",
	}
	for p, body := range contents {
		writeFile(t, p, body)
	}
	wd.start()
	a.tick() // the source reports before the seal, as the seal door requires
	wd.seal()
	wd.sealed()
	wd.flip()
	wd.landed()
	a.tick()
	uk.tick()
	wd.open()

	// A process still in the host's cgroup: nothing is renamed.
	procs := filepath.Join(a.cg, cgroup.ExecName(offHost), "cgroup.procs")
	writeFile(t, procs, "4242\n")
	a.tick()
	if _, err := os.Stat(db); err != nil {
		t.Fatalf("parked with a process left: %v", err)
	}
	if _, ok := wd.reportIf(moves.OfflineRoleSource, "ie-a"); !ok {
		t.Fatal("fixture: no report")
	}
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); r.Phase == moves.OfflinePhaseParked {
		t.Fatalf("reported parked with a process left: %+v", r)
	}
	writeFile(t, procs, "")

	a.tick()
	park := moves.OfflineParkDir(a.dd, offHost, offID, moves.OfflineRoleSource)
	for p, body := range contents {
		switch {
		case strings.HasPrefix(p, a.opt), strings.Contains(p, "neighbour"):
			if got, err := os.ReadFile(p); err != nil || string(got) != body {
				t.Fatalf("%s was touched: %q %v", p, got, err)
			}
			continue
		}
		if _, err := os.Lstat(p); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s is still in place: %v", p, err)
		}
	}
	for rel, body := range map[string]string{
		filepath.Base(db):          "sqlite",
		filepath.Base(db) + "-wal": "wal",
		filepath.Join(filepath.Base(activatorMetaDir(db)), "ltx", "0", "0000000000000001-0000000000000003.ltx"): "ltx",
		filepath.Join(offHost, "instance.key"): "key",
		offHost + ".env":                       "CARLOS_INSTANCE_KEY=k\n",
		offHost + ".config.env":                "SECRET=s\n",
		offHost + ".json":                      `{"programs":[]}`,
	} {
		if got, err := os.ReadFile(filepath.Join(park, rel)); err != nil || string(got) != body {
			t.Fatalf("park/%s = %q %v, want %q", rel, got, err, body)
		}
	}
	r := wd.report(moves.OfflineRoleSource, "ie-a")
	if r.Phase != moves.OfflinePhaseParked || r.LocalDB || r.Files || r.Env || r.Fenced {
		t.Fatalf("parked report: %+v", r)
	}
	if a.svc.Fenced(offHost) {
		t.Fatal("the fence is still up after the park")
	}
	want := map[string]string{
		db: moves.RetiredParked, db + "-wal": moves.RetiredParked, db + "-shm": moves.RetiredAbsent,
		db + "-litestream": moves.RetiredAbsent, activatorMetaDir(db): moves.RetiredParked,
		filepath.Join(a.dd, offHost): moves.RetiredParked, filepath.Join(a.dd, offHost+".env"): moves.RetiredParked,
		filepath.Join(a.dd, offHost+".config.env"): moves.RetiredParked, filepath.Join(a.dd, "programs", offHost+".json"): moves.RetiredParked,
		filepath.Join(a.opt, "live", offHost): moves.RetiredLeft,
	}
	assertRetired(t, r.Retired, want)

	// A second pass changes nothing and reports the same inventory.
	before := r.Retired
	a.tick()
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); !reflect.DeepEqual(r.Retired, before) || r.Phase != moves.OfflinePhaseParked {
		t.Fatalf("second pass: %+v", r.Retired)
	}
	// The row stays held through open.
	if rt := a.row(offHost); rt == nil || rt.HoldOp != offID {
		t.Fatalf("row at open: %+v", rt)
	}

	// finishing: keep no longer holds the row; reconcile removes it and
	// the box reports gone. The parked state stays.
	// The view the reconcile reads is the last moves pass's, so the row
	// goes on the pass after the one that first sees finishing.
	wd.finishing()
	a.tick()
	a.tick()
	if rt := a.row(offHost); rt != nil {
		t.Fatalf("row at finishing: %+v", rt)
	}
	a.pass()
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseGone || r.RowPresent || r.LocalDB {
		t.Fatalf("gone report: %+v", r)
	}
	if got, err := os.ReadFile(filepath.Join(park, filepath.Base(db))); err != nil || string(got) != "sqlite" {
		t.Fatalf("the parked DB: %q %v", got, err)
	}
}

func assertRetired(t *testing.T, got []moves.RetiredPath, want map[string]string) {
	t.Helper()
	seen := map[string]string{}
	for _, rp := range got {
		if _, dup := seen[rp.Path]; dup {
			t.Fatalf("%s listed twice", rp.Path)
		}
		seen[rp.Path] = rp.Action
		if rp.Action == moves.RetiredLeft && rp.Reason == "" {
			t.Fatalf("%s left with no reason", rp.Path)
		}
	}
	for p, action := range want {
		if seen[p] != action {
			t.Fatalf("%s: %q, want %q (inventory %+v)", p, seen[p], action, got)
		}
	}
	if len(seen) != len(want) {
		var extra []string
		for p := range seen {
			if _, ok := want[p]; !ok {
				extra = append(extra, p+"="+seen[p])
			}
		}
		sort.Strings(extra)
		t.Fatalf("unexpected entries: %v", extra)
	}
}

// M3: a path the agent can never rename (here a file in a directory it may
// not write, as a root-owned /data/programs is) is reported left once, with
// its reason, and is not tried again: later passes, an agent restart
// included, neither fence nor rename, and the report stays parked.
func TestOfflineParkLeavesWhatCannotBeRenamedOnce(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("root can rename anything")
	}
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	a.tick()
	db := a.row(offHost).DBPath
	writeFile(t, db, "sqlite")
	programs := filepath.Join(a.dd, "programs")
	writeFile(t, filepath.Join(programs, offHost+".json"), "{}")
	if err := os.Chmod(programs, 0o555); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.Chmod(programs, 0o755) })
	fences := 0
	count := func() {
		inner := a.w.cfg.FenceAs
		a.w.cfg.FenceAs = func(ctx context.Context, host string, gen uint64, owner string) error {
			fences++
			return inner(ctx, host, gen, owner)
		}
	}
	wd.start()
	a.tick()
	wd.seal()
	wd.sealed()
	wd.flip()
	wd.landed()
	a.tick()
	wd.open()
	count()
	before := fences
	a.tick()
	if fences != before+1 {
		t.Fatalf("the first park fenced %d times", fences-before)
	}
	r := wd.report(moves.OfflineRoleSource, "ie-a")
	if r.Phase != moves.OfflinePhaseParked {
		t.Fatalf("phase %s", r.Phase)
	}
	var left moves.RetiredPath
	for _, rp := range r.Retired {
		if rp.Path == filepath.Join(programs, offHost+".json") {
			left = rp
		}
		if rp.Path == db && rp.Action != moves.RetiredParked {
			t.Fatalf("the DB: %+v", rp)
		}
	}
	if left.Action != moves.RetiredLeft || !strings.HasPrefix(left.Reason, leftCannot) {
		t.Fatalf("the unrenameable path: %+v", left)
	}
	for i := 0; i < 2; i++ {
		before = fences
		a.tick()
		if fences != before {
			t.Fatal("a later pass fenced again for a path it cannot rename")
		}
		if again := wd.report(moves.OfflineRoleSource, "ie-a"); again.Phase != moves.OfflinePhaseParked || !reflect.DeepEqual(again.Retired, r.Retired) {
			t.Fatalf("pass %d: %+v", i, again)
		}
	}
	a.restart()
	count()
	before = fences
	a.tick()
	if fences != before || a.svc.Fenced(offHost) {
		t.Fatal("after a restart the pass fenced again for a path it cannot rename")
	}
	if again := wd.report(moves.OfflineRoleSource, "ie-a"); !reflect.DeepEqual(again.Retired, r.Retired) {
		t.Fatalf("after a restart: %+v", again.Retired)
	}
	if _, err := os.Stat(filepath.Join(programs, offHost+".json")); err != nil {
		t.Fatalf("the left path: %v", err)
	}
}

// offline-open on a nested source: the contents of the raw instance
// directory are renamed into the park directory, the directory and its
// view stay, empty, and the inventory names each path.
func TestOfflineOpenParksTheNestedSource(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	a.nested = true
	raw := layout.RawInstanceDir(a.dd, offHost)
	if err := os.MkdirAll(raw, 0o700); err != nil {
		t.Fatal(err)
	}
	// The bind mount at <dd>/instances, as a link in the fixture.
	if err := os.Symlink(filepath.Base(filepath.Dir(raw)), filepath.Join(a.dd, "instances")); err != nil {
		t.Fatal(err)
	}
	a.tick()
	rt := a.row(offHost)
	if rt == nil || !provision.NestedPaths(a.dd, *rt) {
		t.Fatalf("fixture: not a nested row: %+v", rt)
	}
	contents := map[string]string{
		offHost + ".db":     "sqlite",
		offHost + ".db-shm": "shm",
		offHost + ".env":    "CARLOS_INSTANCE_KEY=k\n",
		"instance.key":      "key",
		"uploads/a.png":     "png",
	}
	for rel, body := range contents {
		writeFile(t, filepath.Join(raw, rel), body)
	}
	wd.start()
	a.tick()
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); !r.LocalDB || !r.Files || !r.Env {
		t.Fatalf("nested report: %+v", r)
	}
	wd.seal()
	wd.sealed()
	wd.flip()
	wd.landed()
	wd.open()
	a.tick()
	park := moves.OfflineParkDir(a.dd, offHost, offID, moves.OfflineRoleSource)
	for rel, body := range contents {
		if got, err := os.ReadFile(filepath.Join(park, nestedParkPrefix, rel)); err != nil || string(got) != body {
			t.Fatalf("park/instance/%s = %q %v", rel, got, err)
		}
	}
	ents, err := os.ReadDir(raw)
	if err != nil || len(ents) != 0 {
		t.Fatalf("the instance directory: %v %v, want it there and empty", ents, err)
	}
	r := wd.report(moves.OfflineRoleSource, "ie-a")
	if r.Phase != moves.OfflinePhaseParked || r.LocalDB || r.Files || r.Env {
		t.Fatalf("parked report: %+v", r)
	}
	want := map[string]string{
		filepath.Join(raw, offHost+".db"):                          moves.RetiredParked,
		filepath.Join(raw, offHost+".db-wal"):                      moves.RetiredAbsent,
		filepath.Join(raw, offHost+".db-shm"):                      moves.RetiredParked,
		filepath.Join(raw, offHost+".db-litestream"):               moves.RetiredAbsent,
		filepath.Join(raw, ".activator-"+offHost+".db-litestream"): moves.RetiredAbsent,
		filepath.Join(raw, offHost+".env"):                         moves.RetiredParked,
		filepath.Join(raw, offHost+".config.env"):                  moves.RetiredAbsent,
		filepath.Join(raw, "instance.key"):                         moves.RetiredParked,
		filepath.Join(raw, "uploads"):                              moves.RetiredParked,
		filepath.Join(a.dd, "programs", offHost+".json"):           moves.RetiredAbsent,
		provision.InstanceDir(a.dd, offHost):                       moves.RetiredLeft,
		raw:                                                        moves.RetiredLeft,
		filepath.Join(a.opt, "live", offHost):                      moves.RetiredAbsent,
	}
	assertRetired(t, r.Retired, want)
	// Re-run: the same inventory, from what is parked.
	a.tick()
	if again := wd.report(moves.OfflineRoleSource, "ie-a"); !reflect.DeepEqual(again.Retired, r.Retired) {
		t.Fatalf("second pass:\n %+v\n %+v", again.Retired, r.Retired)
	}
}

// --- the dispatcher and the capability --------------------------------------

// An offline record reaches offlineRound with Phase 1 off, ahead of the
// Phase 1 skip and the lost-box gate: no host claim is read. A
// Phase-1-shaped record with Phase 1 off is still logged and skipped.
func TestOfflineDispatchPrecedesThePhase1Skip(t *testing.T) {
	wd := newOffWorld(t)
	wd.create("other.example.com")
	a := wd.box("ie-a", "", nil)
	a.tick()
	wd.start()
	// A Phase 1 move on the other host (its id is what makes it Phase 1).
	if _, err := provision.SetRecordMove(context.Background(), wd.s, wd.acct, "app", "other.example.com", func(rec *provision.Record) error {
		rec.Move = &provision.MoveState{ID: offID2, From: "", To: "uk", State: provision.MovePreparing, RequestedAt: wd.clock()}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	a.tick()
	if _, ok := wd.reportIf(moves.OfflineRoleSource, "ie-a"); !ok {
		t.Fatal("the offline record never reached offlineRound")
	}
	if a.w.lost != nil {
		t.Fatalf("the lost-box gate ran: %v", a.w.lost)
	}
	if _, ok := a.w.notices.Load("other.example.com#switch"); !ok {
		t.Fatal("the Phase 1 record was not logged as waiting on the switch")
	}
	if rt := a.row("other.example.com"); rt == nil || rt.HoldOp != "" || rt.Status != "ready" {
		t.Fatalf("the Phase 1 record's row was touched: %+v", rt)
	}
	keys, err := wd.s.List(context.Background(), "control/offline-moves/other.example.com/")
	if err != nil || len(keys) != 0 {
		t.Fatalf("the Phase 1 record was reported as offline: %v %v", keys, err)
	}
}

// The capability is the box's own witness: published with hibernation
// configured and both switches unset, withheld otherwise. A box without it
// runs no offline step and installs no offline hook: its row is never
// held, nothing is reported, and one notice is logged. The guard mode is
// not an input (platform-1e's ruling of 2026-10-20).
func TestOfflineCapabilityAndAnIncapableBox(t *testing.T) {
	env := func(kv ...string) func(string) string {
		m := map[string]string{"CARLOS_HIBERNATE_BUCKET": "data"}
		for i := 0; i+1 < len(kv); i += 2 {
			m[kv[i]] = kv[i+1]
		}
		return func(k string) string { return m[k] }
	}
	// The published capability is the armed worker's own answer (M5).
	prev := armedMovesWorker.Load()
	t.Cleanup(func() { armedMovesWorker.Store(prev) })
	armedMovesWorker.Store(nil)
	if armedOfflineCapable() {
		t.Fatal("capable with no worker armed")
	}
	if got := withOfflineCapability(edgesFileCapabilities(env()), armedOfflineCapable()); len(got) != 0 {
		t.Fatalf("no worker armed: %v", got)
	}
	wdCap := newOffWorld(t)
	armed := wdCap.box("armed", "", nil)
	armedMovesWorker.Store(armed.w)
	if !armedOfflineCapable() {
		t.Fatal("a capable armed worker does not publish")
	}
	if got := withOfflineCapability(edgesFileCapabilities(env()), armedOfflineCapable()); !slices.Equal(got, []string{capOfflineMove1, capOfflineStage1}) {
		t.Fatalf("capable box: %v", got)
	}
	base := edgesFileCapabilities(env("CARLOS_REGION_HA", "1"))
	if got := withOfflineCapability(base, true); !slices.Equal(got, []string{capRegionHA1, capOfflineMove1, capOfflineStage1}) || !slices.Equal(base, []string{capRegionHA1}) {
		t.Fatalf("region HA too: %v (base %v)", got, base)
	}
	for _, kv := range [][]string{
		{"CARLOS_HIBERNATE_BUCKET", ""},
		{"CARLOS_HOST_CLAIM", "1"},
		{"CARLOS_HOST_CLAIM", "0"}, // set at all
		{moves.SwitchEnv, "on"},
		{moves.SwitchEnv, "off"}, // set at all
	} {
		if offlineCapable(env(kv...), hibernationConfigured(env(kv...))) {
			t.Fatalf("%s=%q is capable", kv[0], kv[1])
		}
		incapable := wdCap.box("incapable", "", map[string]string{kv[0]: kv[1]})
		armedMovesWorker.Store(incapable.w)
		if got := withOfflineCapability(edgesFileCapabilities(env(kv...)), armedOfflineCapable()); slices.Contains(got, capOfflineMove1) || slices.Contains(got, capOfflineStage1) {
			t.Fatalf("%s=%q publishes: %v", kv[0], kv[1], got)
		}
	}
	armedMovesWorker.Store(prev)

	wd := newOffWorld(t)
	a := wd.box("ie-a", "", map[string]string{"CARLOS_HOST_CLAIM": "1"})
	if a.w.cfg.OfflineCapable {
		t.Fatal("armOffline: a claims box is capable")
	}
	a.tick()
	wd.start()
	a.tick()
	if rt := a.row(offHost); rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
		t.Fatalf("an incapable box held the row: %+v", rt)
	}
	if a.svc.Fenced(offHost) {
		t.Fatal("an incapable box fenced")
	}
	if _, ok := wd.reportIf(moves.OfflineRoleSource, "ie-a"); ok {
		t.Fatal("an incapable box reported")
	}
	if _, ok := a.w.notices.Load(offHost + "#offline"); !ok {
		t.Fatal("no notice from an incapable box")
	}
}

// --- dormant ---------------------------------------------------------------

// flatLayout is a reconcile config's layout read, always flat.
func flatLayout(context.Context, release.Store) (string, error) { return release.LayoutModeFlat, nil }

// dormantWorld is a deployment with no offline record anywhere: offHost at
// rest, gone.example.com (whose record the test deletes), a legacy move and
// a Phase 1 move on records of their own.
func dormantWorld(t *testing.T) *offWorld {
	t.Helper()
	wd := newOffWorld(t)
	for _, h := range []string{"gone.example.com", "legacy.example.com", "phase1.example.com"} {
		wd.create(h)
	}
	wd.moveHost("legacy.example.com", func(rec *provision.Record) error {
		rec.Move = &provision.MoveState{To: "uk", State: provision.MovePreparing}
		return nil
	})
	wd.moveHost("phase1.example.com", func(rec *provision.Record) error {
		rec.Move = &provision.MoveState{ID: offID2, From: "", To: "uk", State: provision.MovePreparing, RequestedAt: wd.clock()}
		return nil
	})
	return wd
}

// dormantBoxes are two home boxes minted from the same records with no
// hooks at all, so their registries start equal: the plain one stands for
// main, the hooked one has a capable worker whose view no listing has read.
func dormantBoxes(t *testing.T, wd *offWorld) (plain, hooked *offBox) {
	t.Helper()
	plain, hooked = wd.box("main", "", nil), wd.box("hooked", "", nil)
	for _, b := range []*offBox{plain, hooked} {
		if err := reconcileSerializedWith(context.Background(), wd.s, b.reg, provision.ReconcileConfig{DataDir: b.dd, Layout: flatLayout}, nil); err != nil {
			t.Fatal(err)
		}
		if b.row("gone.example.com") == nil {
			t.Fatalf("fixture: %s has no row to retire", b.name)
		}
	}
	if hooked.w.cfg.Offline.built() || !hooked.w.cfg.OfflineCapable {
		t.Fatal("fixture: the hooked box's view is built, or the box is not capable")
	}
	return plain, hooked
}

// sameRegistries fails unless the two boxes' registries match, paths and
// creation times aside (each box has its own data directory and clock).
func sameRegistries(t *testing.T, a, b *offBox) {
	t.Helper()
	strip := func(rs []registry.Route) []registry.Route {
		out := append([]registry.Route(nil), rs...)
		for i := range out {
			out[i].DBPath, out[i].Socket, out[i].CreatedAt = "", "", 0
		}
		return out
	}
	ar, err := a.reg.List()
	if err != nil {
		t.Fatal(err)
	}
	br, err := b.reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(strip(ar), strip(br)) {
		t.Fatalf("the registries differ:\n %s %+v\n %s %+v", a.name, ar, b.name, br)
	}
}

// With no offline record anywhere (a legacy and a Phase 1 move included):
// the very first reconcile on a box whose view no listing has read removes
// a retired row exactly as main does; after it the hold answers false for
// every record and keep false for every host; and the edges file differs
// by the one capability string and nothing else.
func TestOfflineDormantIsMain(t *testing.T) {
	ctx := context.Background()
	wd := dormantWorld(t)
	plain, hooked := dormantBoxes(t, wd)
	if err := wd.s.Delete(ctx, provision.RecordKey(wd.acct, "app", "gone.example.com")); err != nil {
		t.Fatal(err)
	}
	for _, b := range []*offBox{plain, hooked} {
		w := b.w
		if b == plain {
			w = nil
		}
		if err := reconcileSerializedWith(ctx, wd.s, b.reg, provision.ReconcileConfig{DataDir: b.dd, Layout: flatLayout}, w); err != nil {
			t.Fatal(err)
		}
		if rt := b.row("gone.example.com"); rt != nil {
			t.Fatalf("%s kept a retired row on its first pass: %+v", b.name, rt)
		}
	}
	sameRegistries(t, plain, hooked)
	v := hooked.w.cfg.Offline
	if !v.built() || v.failed() {
		t.Fatal("the first reconcile did not read the view")
	}
	for _, h := range []string{offHost, "gone.example.com", "legacy.example.com", "phase1.example.com", "never.example.com"} {
		if v.keep(h) {
			t.Fatalf("keep(%s)", h)
		}
	}
	recs, err := provision.ListAllRecords(ctx, wd.s)
	if err != nil {
		t.Fatal(err)
	}
	if len(recs) != 3 {
		t.Fatalf("fixture: %d records", len(recs))
	}
	for _, rec := range recs {
		if _, _, ok := offlineHold(hooked.w.cfg)(rec); ok {
			t.Fatalf("held %s", rec.Host)
		}
	}
	// A further pass changes nothing on either box.
	for _, b := range []*offBox{plain, hooked} {
		w := b.w
		if b == plain {
			w = nil
		}
		if err := reconcileSerializedWith(ctx, wd.s, b.reg, provision.ReconcileConfig{DataDir: b.dd, Layout: flatLayout}, w); err != nil {
			t.Fatal(err)
		}
	}
	sameRegistries(t, plain, hooked)

	// The edges file: the one string, and no headroom with it.
	var last1, last2 []byte
	none := func(string) string { return "" }
	plainFile := edgesFile{Box: "plain", Placement: "", Capabilities: withOfflineCapability(edgesFileCapabilities(none), false)}
	capable := edgesFile{Box: "plain", Placement: "", Capabilities: withOfflineCapability(edgesFileCapabilities(none), true)}
	if headroomCapabilities(capable.Capabilities) != nil {
		t.Fatal("offline-move-1 asks for headroom")
	}
	s1, s2 := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
	if err := edgesFileWrite(ctx, s1, plain.reg, plainFile, "1.2.3.4", &last1, false); err != nil {
		t.Fatal(err)
	}
	if err := edgesFileWrite(ctx, s2, plain.reg, capable, "1.2.3.4", &last2, false); err != nil {
		t.Fatal(err)
	}
	b1, _, _ := s1.Get(ctx, edgesFileKey("plain"))
	b2, _, _ := s2.Get(ctx, edgesFileKey("plain"))
	var m1, m2 map[string]any
	if json.Unmarshal(b1, &m1) != nil || json.Unmarshal(b2, &m2) != nil {
		t.Fatal("edges files do not parse")
	}
	if caps, _ := m2["capabilities"].([]any); len(caps) != 2 || caps[0] != capOfflineMove1 || caps[1] != capOfflineStage1 {
		t.Fatalf("capabilities: %v", m2["capabilities"])
	}
	delete(m2, "capabilities")
	delete(m1, "generation")
	delete(m2, "generation")
	if !reflect.DeepEqual(m1, m2) {
		t.Fatalf("the edges file differs beyond the capability:\n %s\n %s", b1, b2)
	}
	if bytes.Contains(b1, []byte("capabilities")) {
		t.Fatal("fixture: the plain file has capabilities")
	}
}

// M1: a failed startup listing keeps every row only until the next
// reconcile, which reads the listing again: the retired row then goes as on
// main, without waiting for a moves pass.
func TestOfflineFailedListingHealsOnTheNextReconcile(t *testing.T) {
	ctx := context.Background()
	wd := dormantWorld(t)
	plain, hooked := dormantBoxes(t, wd)
	v := hooked.w.cfg.Offline
	refreshOfflineView(ctx, failingListStore{Store: wd.s, prefix: "apps/", err: errors.New("listing refused")}, v)
	if !v.built() || !v.failed() || !v.keep(offHost) {
		t.Fatal("a failed startup listing does not keep")
	}
	if err := wd.s.Delete(ctx, provision.RecordKey(wd.acct, "app", "gone.example.com")); err != nil {
		t.Fatal(err)
	}
	for _, b := range []*offBox{plain, hooked} {
		w := b.w
		if b == plain {
			w = nil
		}
		if err := reconcileSerializedWith(ctx, wd.s, b.reg, provision.ReconcileConfig{DataDir: b.dd, Layout: flatLayout}, w); err != nil {
			t.Fatal(err)
		}
	}
	if v.failed() || v.keep("gone.example.com") {
		t.Fatal("the reconcile did not read the listing again")
	}
	if rt := hooked.row("gone.example.com"); rt != nil {
		t.Fatalf("the retired row stayed after the view healed: %+v", rt)
	}
	sameRegistries(t, plain, hooked)
	// While the listing keeps failing the view stays closed.
	refreshOfflineView(ctx, failingListStore{Store: wd.s, prefix: "apps/", err: errors.New("listing refused")}, v)
	if !v.keep(offHost) {
		t.Fatal("a failed listing does not keep")
	}
	unbuilt := newOfflineView(moveCfg{Region: ""}, nil)
	if !unbuilt.keep(offHost) {
		t.Fatal("an unbuilt view keeps nothing")
	}
}

// M4: a record that does not decode is kept by keep alone (its move cannot
// be known), logged, and every other host is judged as usual. The moves
// pass's listing carries the same answer.
func TestOfflineUndecodableRecordIsKept(t *testing.T) {
	ctx := context.Background()
	wd := newOffWorld(t)
	wd.create("broken.example.com")
	if _, err := wd.s.Put(ctx, provision.RecordKey(wd.acct, "app", "broken.example.com"), []byte("{not json"), mustETag(t, wd.s, provision.RecordKey(wd.acct, "app", "broken.example.com"))); err != nil {
		t.Fatal(err)
	}
	a := wd.box("ie-a", "", nil)
	refreshOfflineView(ctx, wd.s, a.w.cfg.Offline)
	v := a.w.cfg.Offline
	if v.failed() || !v.keep("broken.example.com") || v.keep(offHost) {
		t.Fatalf("startup view: failed %v, keep(broken) %v, keep(%s) %v", v.failed(), v.keep("broken.example.com"), offHost, v.keep(offHost))
	}
	fresh := newOfflineView(a.w.cfg, nil)
	a.w.cfg.Offline = fresh
	a.pass()
	if !fresh.keep("broken.example.com") || fresh.keep(offHost) {
		t.Fatal("the moves pass's listing does not keep the undecodable host")
	}
	if _, _, ok := offlineHold(a.w.cfg)(provision.Record{Host: "broken.example.com"}); ok {
		t.Fatal("held a record with no move")
	}
}

func mustETag(t *testing.T, s release.Store, key string) string {
	t.Helper()
	_, etag, err := s.Get(context.Background(), key)
	if err != nil {
		t.Fatal(err)
	}
	return etag
}

// hostSliceCgroup is the slice instancectl's units run in.
func TestOfflineHostSliceIsInstancectlsSlice(t *testing.T) {
	got := hostSliceCgroup("/sys/fs/cgroup", offHost)
	want := filepath.Join("/sys/fs/cgroup", "carlos.slice", "carlos-instances.slice", instancectl.SliceName(offHost))
	if got != want {
		t.Fatalf("hostSliceCgroup = %s, want %s", got, want)
	}
	if filepath.Dir(filepath.Join("/sys/fs/cgroup", filepath.FromSlash(instancectl.ClaimUnitControlGroup(offHost)))) != got {
		t.Fatal("the run unit is not in the slice")
	}
}

// The seal record's writer is create-only: the first body stays, a re-run
// gets it back and writes nothing, and the report writer beside it replaces.
func TestOfflineCreateOnlyKeepsTheFirstWrite(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	key := moves.OfflineReportPrefix(offHost, offID) + "seal.json"
	got, created, err := putOfflineCreateOnly(ctx, s, key, []byte(`{"n":1}`))
	if err != nil || !created || string(got) != `{"n":1}` {
		t.Fatalf("first write: %q %v %v", got, created, err)
	}
	got, created, err = putOfflineCreateOnly(ctx, s, key, []byte(`{"n":2}`))
	if err != nil || created || string(got) != `{"n":1}` {
		t.Fatalf("second write: %q %v %v, want the first body back, nothing written", got, created, err)
	}
	if body, _, err := s.Get(ctx, key); err != nil || string(body) != `{"n":1}` {
		t.Fatalf("stored: %q %v", body, err)
	}
	if _, _, err := putOfflineCreateOnly(ctx, s, "", nil); err == nil {
		t.Fatal("an empty key was written")
	}
}

// keep's table: the source keeps until finishing, the target until the
// move is cleared, a third region never.
func TestOfflineKeep(t *testing.T) {
	rec := func(region, state string) provision.Record {
		return provision.Record{Host: offHost, Region: region, Move: &provision.MoveState{Kind: provision.MoveKindOffline, ID: offID, From: "", To: "uk", State: state}}
	}
	states := []string{provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding,
		provision.OfflineLanded, provision.OfflineOpen, provision.OfflineFinishing, provision.OfflineReverting}
	for _, st := range states {
		recs := []provision.Record{rec("uk", st)}
		src := newOfflineView(moveCfg{Region: ""}, nil)
		src.refresh(recs, nil)
		if got, want := src.keep(offHost), st != provision.OfflineFinishing; got != want {
			t.Fatalf("source at %s: keep %v", st, got)
		}
		tgt := newOfflineView(moveCfg{Region: "uk"}, nil)
		tgt.refresh(recs, nil)
		if !tgt.keep(offHost) {
			t.Fatalf("target at %s: not kept", st)
		}
		other := newOfflineView(moveCfg{Region: "fr"}, nil)
		other.refresh(recs, nil)
		if other.keep(offHost) {
			t.Fatalf("a third region keeps at %s", st)
		}
	}
}

// readCountingStore counts the reads made through it.
type readCountingStore struct {
	release.Store
	mu    sync.Mutex
	reads int
}

func (c *readCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	c.mu.Lock()
	c.reads++
	c.mu.Unlock()
	return c.Store.Get(ctx, key)
}

func (c *readCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
	c.mu.Lock()
	c.reads++
	c.mu.Unlock()
	return c.Store.List(ctx, prefix)
}

// Harness Run 9, finding 7: between moves passes, a source box rewrites its
// report on the re-assert tick while a door judges that report by age
// (stopping to landed), and only then. It never runs the seal itself, it
// reads nothing while no offline move is in flight, and a box that is
// neither a source nor the sealer does nothing.
func TestOfflineReassertRewritesOnlyTheReportsADoorReads(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	uk := wd.box("uk-box", "uk", nil)
	a.tick()
	uk.tick()
	cs := &readCountingStore{Store: wd.s}
	ctx := context.Background()
	if n := a.w.offlineReassert(ctx, cs); n != 0 || cs.reads != 0 {
		t.Fatalf("no move in flight: acted on %d, %d store reads", n, cs.reads)
	}

	wd.start()
	a.tick() // the start door's bump
	uk.tick()
	wd.advance(3 * time.Minute)
	for _, state := range []string{provision.OfflineStopping, provision.OfflineSealing} {
		if state == provision.OfflineSealing {
			wd.seal()
			wd.advance(time.Minute)
		}
		if n := a.w.offlineReassert(ctx, cs); n != 1 {
			t.Fatalf("%s: the source acted on %d records, want 1", state, n)
		}
		if r := wd.report(moves.OfflineRoleSource, "ie-a"); !r.At.Equal(wd.clock()) || !r.Held || !r.Fenced {
			t.Fatalf("%s: the source's report %+v, want rewritten at %v", state, r, wd.clock())
		}
		if n := uk.w.offlineReassert(ctx, cs); n != 0 {
			t.Fatalf("%s: the target acted on %d records", state, n)
		}
	}
	if _, _, err := wd.s.Get(ctx, moves.OfflineSealKey(offHost, offID)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the re-assert ran the seal (err %v)", err)
	}

	// From open on, no door reads a report by age: nothing is rewritten.
	wd.sealed()
	wd.flip()
	wd.landed()
	wd.open()
	a.pass() // the view learns the state the way it always does
	at := wd.report(moves.OfflineRoleSource, "ie-a").At
	wd.advance(time.Minute)
	if n := a.w.offlineReassert(ctx, cs); n != 0 {
		t.Fatalf("open: the source acted on %d records", n)
	}
	if r := wd.report(moves.OfflineRoleSource, "ie-a"); !r.At.Equal(at) {
		t.Fatalf("open: the source's report was rewritten at %v", r.At)
	}
}

// elsewhere, scheduleSkip's second half (harness Run 9, defect 8): the box
// whose region the record does not name, through every state the record
// moves through.
func TestOfflineElsewhereFollowsTheRecordsRegion(t *testing.T) {
	wd := newOffWorld(t)
	a := wd.box("ie-a", "", nil)
	uk := wd.box("uk-box", "uk", nil)
	a.tick()
	uk.tick()
	check := func(when string, ieElsewhere, ukElsewhere bool) {
		t.Helper()
		a.pass()
		uk.pass()
		if got := a.w.cfg.Offline.elsewhere(offHost); got != ieElsewhere {
			t.Fatalf("%s: ie-a elsewhere = %v", when, got)
		}
		if got := uk.w.cfg.Offline.elsewhere(offHost); got != ukElsewhere {
			t.Fatalf("%s: uk-box elsewhere = %v", when, got)
		}
	}
	check("no move", false, false)
	wd.start()
	check("stopping", false, true)
	wd.seal()
	wd.sealed()
	check("sealed", false, true)
	wd.flip()
	check("landing", true, false)
	wd.abort()
	check("reverting", false, true)
}