package provision

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

	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

var ctx = context.Background()

// acctID is the id behind every test's account manifest. Its sqid is always
// computed with domains.Sqid, never hardcoded — the same discipline
// internal/domains/reconcile_test.go follows for the identical reason: a
// literal sqid string would silently stop agreeing with domains.Sqid the
// moment the alphabet or offset ever changed.
const acctID int64 = 2

func dirStore(t *testing.T) *release.DirStore {
	t.Helper()
	return release.NewDirStore(t.TempDir())
}

func regWith(t *testing.T, routes ...registry.Route) *registry.Registry {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	for _, rt := range routes {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	return reg
}

func mustGet(t *testing.T, reg *registry.Registry, host string) *registry.Route {
	t.Helper()
	rt, err := reg.Get(host)
	if err != nil {
		t.Fatal(err)
	}
	if rt == nil {
		t.Fatalf("route %s missing", host)
	}
	return rt
}

func accountManifestBody(id int64, sqid, name string) []byte {
	return []byte(`{"id":` + fmt.Sprint(id) + `,"sqid":"` + sqid + `","name":"` + name +
		`","created_at":"2026-08-02T09:00:00Z"}`)
}

// writeAccountManifest writes a fresh (create-only) account manifest and
// hands back its sqid — the frozen encoding of id, computed by
// domains.Sqid, never a literal.
func writeAccountManifest(t *testing.T, s release.Store, id int64, name string) string {
	t.Helper()
	sqid := domains.Sqid(id)
	if _, err := s.Put(ctx, "accounts/"+sqid+"/account.json", accountManifestBody(id, sqid, name), ""); err != nil {
		t.Fatal(err)
	}
	return sqid
}

// putReplace overwrites an existing object (read-etag-then-put), for tests
// that need to corrupt or repair an object already written by
// writeAccountManifest's create-only Put.
func putReplace(t *testing.T, s release.Store, key string, body []byte) {
	t.Helper()
	_, etag, err := s.Get(ctx, key)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, key, body, etag); err != nil {
		t.Fatal(err)
	}
}

// releaseKeys mints a throwaway release signing pair — the same helper
// internal/domains/reconcile_test.go carries, for the same reason: the
// production flag is signed, so a test that exercises the guard needs both
// halves of a real key.
func releaseKeys(t *testing.T) (ed25519.PrivateKey, ed25519.PublicKey) {
	t.Helper()
	seedB64, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := release.ParseSigningKey(seedB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	pub, err := release.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	return priv, pub
}

func mustJSON(t *testing.T, v any) []byte {
	t.Helper()
	b, err := json.Marshal(v)
	if err != nil {
		t.Fatal(err)
	}
	return b
}

// setup writes a valid account manifest (id acctID) and that account's
// "tito" app instances manifest (domain go.tito.io, pool tito, seed
// ["seed"]) — the fixture every test case below builds records on top of.
func setup(t *testing.T) (*release.DirStore, string) {
	t.Helper()
	s := dirStore(t)
	sqid := writeAccountManifest(t, s, acctID, "Tito")
	if err := PutManifest(ctx, s, sqid, "tito", Manifest{
		Schema: 1, App: "tito", Domain: "go.tito.io", Pool: "tito", Seed: []string{"seed"},
	}); err != nil {
		t.Fatal(err)
	}
	return s, sqid
}

func acmeRecord(sqid string) Record {
	return Record{Schema: 1, Host: "acme.go.tito.io", Slug: "acme", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()}
}

// 1. Mint.
func TestReconcileMints(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	got := *mustGet(t, reg, "acme.go.tito.io")
	got.CreatedAt = 0 // stamped by Upsert; not part of the contract under test
	want := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	}
	if got != want {
		t.Errorf("minted route = %+v, want %+v", got, want)
	}
}

// Every shape the record can now describe, minted in one pass. The four
// cases are the production routes the old vocabulary could not express:
// a console (addr-backed, unit-owned, never hibernating), a whole-app TCP
// tenant, a static site, and a sibling service.
func TestReconcileMintsEveryShape(t *testing.T) {
	s, sqid := setup(t)
	off := false
	on := true
	recs := []Record{
		{Schema: 1, Host: "console.go.tito.io", Slug: "console", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Kind: "instance",
			Addr: "127.0.0.1:8090", Unit: "console.service", CreatedAt: time.Now()},
		{Schema: 1, Host: "whole.go.tito.io", Slug: "whole", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Addr: "127.0.0.1:9443",
			Hibernate: &off, CreatedAt: time.Now()},
		{Schema: 1, Host: "site.go.tito.io", Slug: "site", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Kind: "static",
			StatusNamed: &on, CreatedAt: time.Now()},
		{Schema: 1, Host: "svc.go.tito.io", Slug: "svc", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
			Addr: "127.0.0.1:7000", Environment: "staging", CreatedAt: time.Now()},
		{Schema: 1, Host: "ha.go.tito.io", Slug: "ha", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
			Addrs: []string{"127.0.0.1:7001", "127.0.0.1:7002"}, CreatedAt: time.Now()},
	}
	for _, rec := range recs {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	console := *mustGet(t, reg, "console.go.tito.io")
	console.CreatedAt = 0
	wantConsole := registry.Route{
		Host: "console.go.tito.io", Slug: "console", Kind: "instance",
		Addr: "127.0.0.1:8090", Unit: "console.service",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Provisioned: true, Seed: `["seed"]`, Status: "ready",
	}
	if console != wantConsole {
		t.Errorf("console route = %+v, want %+v", console, wantConsole)
	}

	whole := *mustGet(t, reg, "whole.go.tito.io")
	if whole.Socket != "" || whole.DBPath != "" {
		t.Errorf("an addr-backed tenant derived filesystem paths: socket %q db %q", whole.Socket, whole.DBPath)
	}
	if whole.Hibernate {
		t.Error("an explicitly non-hibernating record minted a hibernating route")
	}

	site := *mustGet(t, reg, "site.go.tito.io")
	if site.Kind != "static" {
		t.Errorf("static route Kind = %q", site.Kind)
	}
	if site.Socket != "" || site.DBPath != "" {
		t.Errorf("a static route derived filesystem paths: socket %q db %q", site.Socket, site.DBPath)
	}
	if site.Hibernate {
		t.Error("a static route was armed for hibernation — it has no process to wake")
	}
	if !site.StatusNamed {
		t.Error("StatusNamed did not reach the row")
	}

	// A sibling service, and the INSERT path for Environment. On the conflict
	// path Upsert deliberately preserves a non-empty environment (`CASE WHEN
	// excluded.environment = ''`), so the insert is the only place a mint's
	// Environment can land at all — and every other assertion in this file
	// would pass just the same if it never did.
	svc := *mustGet(t, reg, "svc.go.tito.io")
	svc.CreatedAt = 0
	wantSvc := registry.Route{
		Host: "svc.go.tito.io", Slug: "svc", Kind: "service", Addr: "127.0.0.1:7000",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Environment: "staging", Provisioned: true, Seed: `["seed"]`, Status: "ready",
	}
	if svc != wantSvc {
		t.Errorf("service route = %+v, want %+v", svc, wantSvc)
	}

	// The routeFor -> reg.Upsert hand-off, pinned end-to-end through a real
	// mint pass: a multi-addr record's Addrs (and its implied default
	// Health) must reach the row Reconcile actually wrote, not just the
	// value routeFor returns in isolation (TestRouteForCarriesAddrsAndHealth
	// covers that half). This is the fix for the hazard Task 2 flagged —
	// SetUpstreams was safe in isolation, but every mint-path Upsert would
	// have blanked it, had routeFor not been taught to carry the columns.
	ha := *mustGet(t, reg, "ha.go.tito.io")
	if ha.Addrs != "127.0.0.1:7001,127.0.0.1:7002" {
		t.Errorf("multi-addr record's Addrs did not reach the registry row: %q", ha.Addrs)
	}
	if ha.Health == "" {
		t.Error("multi-addr record's default Health did not reach the registry row")
	}
}

// 2. Scope.
func TestReconcileScope(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)

	// Wrong region entirely.
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "us", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("minted for a foreign region: %+v %v", rt, err)
	}

	// Right region, but this box serves no pools (nil = the shared "" pool
	// only) — "tito" is not among them.
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("minted for an unserved pool: %+v %v", rt, err)
	}

	// A record with empty pool+region mints under a zero-value cfg (the
	// unconstrained region, the shared pool).
	unscoped := Record{Schema: 1, Host: "shared.go.tito.io", Slug: "shared", App: "tito",
		Account: sqid, Channel: "stable", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, unscoped); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, ReconcileConfig{}); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "shared.go.tito.io")
}

// 2b. Fleet placement rides the same Region filter, string-equal, unchanged.
//
// TestReconcileFleetRegionMatchesByStringEquality pins the customer-fleets
// plan's Task 5 Step 1: a record placed on a fleet (Region: "fleet/pi") is
// adopted by a reconcile pass configured for that exact fleet region, and
// left alone by a pass configured for a different (platform) region —
// reconcile.go:125's `rec.Region != cfg.Region` is plain string equality,
// so "fleet/pi" needs no new grammar to be recognized: it's simply another
// opaque region string, exactly like "ie" or "us" already are. Exercises
// the real Reconcile entry point (not a reimplementation of the filter) so
// a future refactor of the scope check can't silently narrow it.
func TestReconcileFleetRegionMatchesByStringEquality(t *testing.T) {
	s, sqid := setup(t)
	rec := Record{Schema: 1, Host: "acme.go.tito.io", Slug: "acme", App: "tito",
		Account: sqid, Pool: "tito", Region: "fleet/pi", Channel: "stable", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)

	// A pass configured for a platform region must not adopt a fleet-placed
	// record — "eu" != "fleet/pi", same as any other region mismatch.
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "eu", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("minted for the wrong region (eu != fleet/pi): %+v %v", rt, err)
	}

	// The fleet's own box — configured with the exact "fleet/pi" region —
	// adopts it with zero code change: the same Region == cfg.Region check
	// every platform region already uses.
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "fleet/pi", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	got := mustGet(t, reg, "acme.go.tito.io")
	if got.Region != "fleet/pi" {
		t.Errorf("adopted route Region = %q, want fleet/pi", got.Region)
	}
}

// 3. Idempotent.
func TestReconcileIdempotent(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(before, after) {
		t.Errorf("second pass changed the registry:\nbefore %+v\nafter  %+v", before, after)
	}
}

// 4. Converge seed.
func TestReconcileConvergesSeed(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")

	if err := PutManifest(ctx, s, sqid, "tito", Manifest{
		Schema: 1, App: "tito", Domain: "go.tito.io", Pool: "tito", Seed: []string{"seed", "-v"},
	}); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after.Seed != `["seed","-v"]` {
		t.Errorf("seed = %q, want the converged manifest seed", after.Seed)
	}
	after.Seed = before.Seed
	if after != before {
		t.Errorf("converge touched more than Seed: before %+v after %+v", before, after)
	}
}

// Issue #130, as it happened in production on 2026-08-09: a record created
// on stable, deleted and recreated on edge, and the route row left on
// stable — a route pointing at a channel with no pointer, so every wake
// failed "no adoptable head" while carlos channels cheerfully showed edge
// promoted. Reconcile created from a record and never converged to one.
func TestReconcileConvergesChannel(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")

	rec.Channel = "edge"
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after.Channel != "edge" {
		t.Errorf("channel = %q, want edge — the record moved and the row did not", after.Channel)
	}
	after.Channel = before.Channel
	if after != before {
		t.Errorf("converge touched more than Channel: before %+v after %+v", before, after)
	}

	// The fleet runs on canary/* channels today (pinned in
	// internal/release/laddercheck_test.go); a converge that could not carry
	// one would stop deploys.
	rec.Channel = "canary/rehearsal"
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "acme.go.tito.io").Channel; got != "canary/rehearsal" {
		t.Errorf("channel = %q, want canary/rehearsal", got)
	}
}

// The converge guard is load-bearing: an empty Channel means a malformed
// record, not "clear the channel" — an empty channel would pin the route to
// nothing, exactly the #130 failure mode. A route already provisioned on a
// real channel must keep it when its record goes channel-less, not have it
// blanked.
func TestReconcileConvergeGuardEmptyChannelLeavesRouteAlone(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")
	if before.Channel != "stable" {
		t.Fatalf("fixture's route didn't channel as expected: %+v", before)
	}

	rec.Channel = ""
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("channel-less record changed the row: before %+v after %+v", before, after)
	}
}

func TestReconcileConvergesEnvironment(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	rec.Environment = "staging"
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "acme.go.tito.io").Environment; got != "staging" {
		t.Errorf("environment = %q, want staging", got)
	}

	// Clearing must reach the row too. Upsert deliberately PRESERVES a
	// non-empty environment (it cannot tell "untouched" from "cleared"), so
	// the clear has to go through the narrow setter — which is exactly why
	// this converge does not re-Upsert.
	rec.Environment = ""
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "acme.go.tito.io").Environment; got != "" {
		t.Errorf("environment = %q, want it cleared", got)
	}
}

// TestReconcileConvergesSteering: a record's steering opt-in changing after
// mint converges onto the owned row within one pass, exactly as Channel and
// Environment do — the console's PATCH door (Task 3) writes the record, and
// this is the pass that has to notice.
func TestReconcileConvergesSteering(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")
	if before.Steering != "" {
		t.Fatalf("minted Steering = %q, want empty", before.Steering)
	}

	rec.Steering = "latency"
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after.Steering != "latency" {
		t.Errorf("steering = %q, want latency", after.Steering)
	}
	// The self-review question this test also answers: an edited record
	// converges ONLY its steering column, through SetSteering's narrow
	// UPDATE — not a full-row Upsert that could disturb anything else.
	after.Steering = before.Steering
	if after != before {
		t.Errorf("converging steering touched another column:\n before %+v\n after  %+v", before, after)
	}

	// Clearing must reach the row too — same shape as Environment's clear:
	// Upsert never runs on the owned branch, so only the narrow setter can
	// move this column back to "".
	rec.Steering = ""
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "acme.go.tito.io").Steering; got != "" {
		t.Errorf("steering = %q, want it cleared", got)
	}
}

func TestReconcileConvergesGate(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")
	if before.Gate != "" {
		t.Fatalf("minted Gate = %q, want open", before.Gate)
	}

	// From here on the log is captured: the converge line names the STATE,
	// never the secret. A bcrypt hash is not a password, but it is the
	// material an offline cracker works on, and an operations log is copied,
	// shipped and pasted far more freely than the registry row is.
	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	const hash = "$2a$10$N9qo8uLOickgx2ZMRZoMyeIjZAgcfl7p92ldGxad68LJZdL17lhWy"
	rec.Gate = hash
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(logged.String(), "converged acme.go.tito.io gate open -> set") {
		t.Errorf("no open -> set line logged; log was:\n%s", logged.String())
	}
	if strings.Contains(logged.String(), hash) {
		t.Errorf("the hash reached the log:\n%s", logged.String())
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after.Gate != hash {
		t.Errorf("gate = %q, want the hash", after.Gate)
	}
	after.Gate = before.Gate
	if after != before {
		t.Errorf("converging gate touched another column:\n before %+v\n after  %+v", before, after)
	}

	// A record with no gate field is an OPEN host — it is a string, not a
	// pointer: absent and "" are the same fact, and clearing must reach the
	// row through SetGate.
	rec.Gate = ""
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	logged.Reset()
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "acme.go.tito.io").Gate; got != "" {
		t.Errorf("gate = %q, want it cleared", got)
	}
	if !strings.Contains(logged.String(), "converged acme.go.tito.io gate set -> open") {
		t.Errorf("no set -> open line logged; log was:\n%s", logged.String())
	}
	if strings.Contains(logged.String(), hash) {
		t.Errorf("the hash reached the log:\n%s", logged.String())
	}
}

// An explicit Hibernate converges; an absent one never does. Absent means
// "the record does not speak to this", and a pass that stamped its
// derivation over an operator's deliberate SetHibernate would be the silent
// stomp this design exists to end.
func TestReconcileConvergesHibernateOnlyWhenExplicit(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHibernate("acme.go.tito.io", false); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if mustGet(t, reg, "acme.go.tito.io").Hibernate {
		t.Error("an absent Hibernate re-asserted its derivation over an operator's setting")
	}

	on := true
	rec.Hibernate = &on
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, "acme.go.tito.io").Hibernate {
		t.Error("an explicit Hibernate did not converge")
	}
}

// Converging hibernation OFF drops the route's replication rung with it.
// This is the quietest of the three doors that flip the column — no operator
// types it, a record edit somewhere else does — and it is the one that made
// `carlos route --replication`'s rail point-in-time: the rung is read only on
// the activator's wake path, so the moment this pass stops a route sleeping,
// box-wide /etc/litestream.yml owns that database and a stored `daily` is a
// column `carlos routes` prints and nothing honours (discussion #10).
//
// A record cannot express a rung — it is box-scoped, like caps and tunnel —
// so this pass can only drop one, never restore it. That is why the drop is
// logged beside the flip rather than left to be discovered.
func TestReconcileConvergingHibernateOffClearsTheReplicationRung(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	on := true
	rec.Hibernate = &on
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, "acme.go.tito.io").Hibernate {
		t.Fatal("setup: the minted route does not hibernate")
	}
	// The operator's own write, on the box, through the only door there is.
	if err := reg.SetReplication("acme.go.tito.io", registry.ReplicationDaily); err != nil {
		t.Fatal(err)
	}

	off := false
	rec.Hibernate = &off
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	rt := mustGet(t, reg, "acme.go.tito.io")
	if rt.Hibernate {
		t.Fatal("setup: hibernate did not converge off")
	}
	if rt.Replication != "" {
		t.Errorf("replication = %q after the record stopped the route hibernating — the column now names an interval nothing reads", rt.Replication)
	}
}

// nestRow moves a minted flat row onto the nested layout the way the layout
// pass's compare-and-swap does, so a test can meet a nested row without the
// pass itself.
func nestRow(t *testing.T, reg *registry.Registry, dataDir, host string) {
	t.Helper()
	if err := reg.SetInstancePaths(host, DefaultDBPath(dataDir, host), DefaultSocket(host),
		NestedDBPath(dataDir, host), NestedSocket(host)); err != nil {
		t.Fatal(err)
	}
}

// A record that stops a nested row hibernating is not applied: the row would
// leave the activator's management with its files under the instance tree,
// and the revert pass only quiesces rows the activator manages.
func TestReconcileKeepsANestedRowFlagged(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	on := true
	rec.Hibernate = &on
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	nestRow(t, reg, cfg.DataDir, rec.Host)

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	off := false
	rec.Hibernate = &off
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, rec.Host).Hibernate {
		t.Fatal("the record cleared hibernate on a nested row")
	}
	want := "provision: acme.go.tito.io: record asks to clear hibernate but the row's paths are nested; revert the layout first"
	if !strings.Contains(logged.String(), want) {
		t.Errorf("no refusal logged; want %q in:\n%s", want, logged.String())
	}
	if strings.Contains(logged.String(), "converged acme.go.tito.io hibernate") {
		t.Errorf("a converge was logged over the refusal:\n%s", logged.String())
	}

	// A flat row still converges off: the refusal is by path shape.
	flat := acmeRecord(sqid)
	flat.Host, flat.Slug = "flat.go.tito.io", "flat"
	flat.Hibernate = &on
	if err := CreateRecord(ctx, s, flat); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	flat.Hibernate = &off
	putReplace(t, s, RecordKey(sqid, "tito", flat.Host), mustJSON(t, flat))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if mustGet(t, reg, flat.Host).Hibernate {
		t.Error("a flat row's hibernate did not converge off")
	}
}

// A flat row with a layout journal is mid-move, and clearing its flag
// would strand it the same way, so the record is not applied and the same
// approved line is logged.
func TestReconcileKeepsAFlatRowFlaggedWhileAJournalExists(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	on := true
	rec.Hibernate = &on
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: t.TempDir()}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	journal := LayoutJournalPath(cfg.DataDir, rec.Host)
	if err := os.MkdirAll(filepath.Dir(journal), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(journal, []byte("{}"), 0o600); err != nil {
		t.Fatal(err)
	}

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	off := false
	rec.Hibernate = &off
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if NestedPaths(cfg.DataDir, *mustGet(t, reg, rec.Host)) {
		t.Fatal("setup: the row is nested; this test is about a flat row")
	}
	if !mustGet(t, reg, rec.Host).Hibernate {
		t.Fatal("the record cleared hibernate on a row with a layout journal")
	}
	want := "provision: acme.go.tito.io: record asks to clear hibernate but the row's paths are nested; revert the layout first"
	if !strings.Contains(logged.String(), want) {
		t.Errorf("no refusal logged; want %q in:\n%s", want, logged.String())
	}
}

// A nested row's paths differ from the record's flat derivation by design,
// so the drift log names the layout rather than a shape mismatch that
// points the operator at `carlos remove`.
func TestReconcileNestedRowIsNotShapeDrift(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	on := true
	rec.Hibernate = &on
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	nestRow(t, reg, cfg.DataDir, rec.Host)

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	want := fmt.Sprintf("provision: acme.go.tito.io: row is on the nested layout (db %q socket %q); the record's shared paths are not drift",
		NestedDBPath(cfg.DataDir, rec.Host), NestedSocket(rec.Host))
	if !strings.Contains(logged.String(), want) {
		t.Errorf("want %q in:\n%s", want, logged.String())
	}
	if strings.Contains(logged.String(), "different route SHAPE") {
		t.Errorf("a nested row was reported as shape drift:\n%s", logged.String())
	}
}

// StatusNamed is converged, so it carries Hibernate's tri-state for
// Hibernate's reason: explicit true reaches the row, explicit false CLEARS
// it, and absent is not an answer at all.
func TestReconcileConvergesStatusNamedWhenExplicit(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if mustGet(t, reg, "acme.go.tito.io").StatusNamed {
		t.Fatal("a record with no status_named minted an opted-in route")
	}

	on := true
	rec.StatusNamed = &on
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, "acme.go.tito.io").StatusNamed {
		t.Error("an explicit status_named:true did not converge onto the row")
	}

	off := false
	rec.StatusNamed = &off
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if mustGet(t, reg, "acme.go.tito.io").StatusNamed {
		t.Error("an explicit status_named:false did not clear the row's opt-in")
	}
}

// The whole reason StatusNamed is a pointer: an operator's `carlos add
// -status-named` opt-in survives every pass of a record that never mentions
// the field. A plain bool would read absent as an explicit false and quietly
// pull the app's name off the public status page.
func TestReconcileAbsentStatusNamedDoesNotStompOptIn(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetStatusNamed("acme.go.tito.io", true); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, "acme.go.tito.io").StatusNamed {
		t.Error("an absent StatusNamed stamped its default over an operator's opt-in")
	}
}

// haRecord is the multi-upstream fixture: a service route fronting two
// processes, the shape the HA design (issue #152) added Addrs/Health for.
func haRecord(sqid string) Record {
	return Record{Schema: 1, Host: "ha.go.tito.io", Slug: "ha", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
		Addrs: []string{"127.0.0.1:7001", "127.0.0.1:7002"}, CreatedAt: time.Now()}
}

// TestReconcileConvergesUpstreams: a record whose upstream set (or probe
// config) changes AFTER the route was minted converges onto the owned row
// within one pass, exactly as Channel and Environment do.
//
// This is the drift Task 7's review found: registry.SetUpstreams existed with
// no non-test caller and the owned-row branch compared Kind/Addr/Unit/Socket/
// DBPath only, so a record that gained, lost or reordered an upstream kept
// serving the set it was minted with forever — silently, since the drift log
// never looked at those columns either. The console's PATCH door writes
// exactly this record field, so without this converge that door would move
// the bucket and nothing else.
func TestReconcileConvergesUpstreams(t *testing.T) {
	s, sqid := setup(t)
	rec := haRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "ha.go.tito.io")
	if before.Addrs != "127.0.0.1:7001,127.0.0.1:7002" {
		t.Fatalf("minted Addrs = %q, want the pair", before.Addrs)
	}

	// One upstream replaced: the row must follow the record.
	rec.Addrs = []string{"127.0.0.1:7001", "127.0.0.1:7003"}
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "ha.go.tito.io")
	if after.Addrs != "127.0.0.1:7001,127.0.0.1:7003" {
		t.Errorf("Addrs = %q, want the record's new set — the record moved and the row did not", after.Addrs)
	}
	after.Addrs = before.Addrs
	if after != before {
		t.Errorf("the upstream converge touched more than Addrs: before %+v after %+v", before, after)
	}

	// ORDER is shape-significant (shapeMismatch compares the joined column
	// verbatim), so a reordered record is a converge, not a no-op.
	rec.Addrs = []string{"127.0.0.1:7003", "127.0.0.1:7001"}
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "ha.go.tito.io").Addrs; got != "127.0.0.1:7003,127.0.0.1:7001" {
		t.Errorf("Addrs = %q, want the record's order", got)
	}

	// The probe config converges through the same setter, and lands
	// fully-defaulted — an un-defaulted interval_ms of 0 reaching the edge
	// would be a tight-loop probe against a customer upstream.
	rec.Health = &Health{Path: "/healthz"}
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	var probe Health
	if err := json.Unmarshal([]byte(mustGet(t, reg, "ha.go.tito.io").Health), &probe); err != nil {
		t.Fatalf("row Health is not probe JSON: %v", err)
	}
	if probe.Path != "/healthz" || probe.IntervalMS != 2000 || probe.Status != 200 {
		t.Errorf("converged probe = %+v, want /healthz with the platform defaults filled in", probe)
	}

	// A pass over an unchanged record writes nothing further: converge is
	// idempotent, not a write every tick.
	steady := *mustGet(t, reg, "ha.go.tito.io")
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if again := *mustGet(t, reg, "ha.go.tito.io"); again != steady {
		t.Errorf("a second pass over an unchanged record moved the row: %+v -> %+v", steady, again)
	}
}

// TestReconcileConvergesLegacyAddrToAddrs is the migration this whole feature
// exists to make possible: a tenant that has been serving on ONE addr since
// before Addrs existed (kass, keymail, the consoles) gains a second upstream
// through the console's PATCH door, which rewrites the record from Addr to
// Addrs. The row must end up agreeing with the record on ALL THREE upstream
// columns.
//
// Found in review, 2026-08-12: converging addrs while leaving the row's addr
// standing cost three separate things — the pass logged "record describes a
// different route SHAPE ... carlos remove" every tick forever, the row could
// never be re-adopted (shapeMismatch compares addr), and clearing the set
// later would resurrect the decommissioned upstream through AddrList's fold.
func TestReconcileConvergesLegacyAddrToAddrs(t *testing.T) {
	s, sqid := setup(t)
	rec := Record{Schema: 1, Host: "whole.go.tito.io", Slug: "whole", App: "tito", Account: sqid,
		Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
		Addr: "127.0.0.1:9443", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, "whole.go.tito.io"); got.Addr != "127.0.0.1:9443" || got.Addrs != "" {
		t.Fatalf("legacy mint = addr %q addrs %q, want the single addr alone", got.Addr, got.Addrs)
	}

	// What the PATCH door writes: the set stated whole, the legacy spelling
	// retired.
	rec.Addr = ""
	rec.Addrs = []string{"127.0.0.1:9443", "127.0.0.1:9444"}
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "whole.go.tito.io")
	if after.Addr != "" {
		t.Errorf("row Addr = %q after the migration, want it retired — AddrList would fold it back in", after.Addr)
	}
	if after.Addrs != "127.0.0.1:9443,127.0.0.1:9444" {
		t.Errorf("row Addrs = %q, want the record's set", after.Addrs)
	}
	if after.Health == "" {
		t.Error("a multi-upstream row converged without a probe — the edge cannot tell which upstream is dark")
	}
	if list := after.AddrList(); len(list) != 2 {
		t.Errorf("AddrList = %v, want exactly the two upstreams (a stale addr folded back in is live traffic to a dead backend)", list)
	}

	// The row now IS the route the record describes: no drift, and adoptable
	// (shapeMismatch is the adoption gate AND the drift test — the two failures
	// the un-cleared addr caused, asserted at their source).
	seed := seedJSON(&Manifest{Seed: []string{"seed"}}) // setup()'s manifest
	if reason := shapeMismatch(after, RouteFor(rec, cfg.DataDir, seed, false, nil)); reason != "" {
		t.Errorf("converged row still diverges from its record (%s) — the pass would log this every tick and refuse re-adoption", reason)
	}

	// Clearing the set on a route that no longer has a legacy addr to fall
	// back on: the columns empty out together, and nothing resurrects.
	rec.Addrs = nil
	rec.Kind = "instance" // a service with no upstream is refused by ValidateShape
	putReplace(t, s, RecordKey(sqid, "tito", rec.Host), mustJSON(t, rec))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	cleared := *mustGet(t, reg, "whole.go.tito.io")
	if cleared.Addr != "" || cleared.Addrs != "" || cleared.Health != "" {
		t.Errorf("cleared row = addr %q addrs %q health %q, want all three empty", cleared.Addr, cleared.Addrs, cleared.Health)
	}
	if cleared.AddrList() != nil {
		t.Errorf("AddrList = %v after clearing, want nil — a resurrected upstream is traffic to a decommissioned backend", cleared.AddrList())
	}
}

// TestReconcileLeavesLegacySingleAddrUpstreamsAlone is the additivity guard
// on the converge above: every record in a production bucket today is
// single-Addr or socket-backed, carries no addrs and no probe, and must
// converge to the same empty columns it has always had — never to a
// one-element Addrs the fold-back (AddrsOrDefault) would produce.
func TestReconcileLeavesLegacySingleAddrUpstreamsAlone(t *testing.T) {
	s, sqid := setup(t)
	for _, rec := range []Record{
		acmeRecord(sqid), // socket-backed instance, no addr at all
		{Schema: 1, Host: "whole.go.tito.io", Slug: "whole", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
			Addr: "127.0.0.1:9443", CreatedAt: time.Now()},
	} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := map[string]registry.Route{}
	for _, h := range []string{"acme.go.tito.io", "whole.go.tito.io"} {
		before[h] = *mustGet(t, reg, h)
		if before[h].Addrs != "" || before[h].Health != "" {
			t.Fatalf("%s minted with upstream columns set: %+v", h, before[h])
		}
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	for h, want := range before {
		if got := *mustGet(t, reg, h); got != want {
			t.Errorf("%s moved on a converge pass: %+v -> %+v", h, want, got)
		}
	}
}

// 5. Conflict.
func TestReconcileLeavesConflictAlone(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{Host: "acme.go.tito.io", Kind: "instance", App: "other",
		Status: "ready", Addr: "127.0.0.1:9000"}
	reg := regWith(t, handAdded)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	before := *mustGet(t, reg, "acme.go.tito.io")
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if before != after {
		t.Errorf("conflicting hand-added route changed: before %+v after %+v", before, after)
	}
	if n, err := reg.List(); err != nil || len(n) != 1 {
		t.Errorf("pass minted something alongside the conflict: %+v %v", n, err)
	}
}

// 6. Remove.
func TestReconcileRemoves(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{Host: "neighbor.go.tito.io", Kind: "instance", App: "other",
		Status: "ready", Addr: "127.0.0.1:9001"}
	foreign := registry.Route{Host: "us.go.tito.io", Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "us", Pool: "tito", Account: sqid}
	reg := regWith(t, handAdded, foreign)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io") // minted first

	if err := DeleteRecord(ctx, s, sqid, "tito", "acme.go.tito.io"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("retired route survived: %+v %v", rt, err)
	}
	mustGet(t, reg, "neighbor.go.tito.io") // hand-added, never provisioned: survives
	mustGet(t, reg, "us.go.tito.io")       // provisioned but out of this box's region: survives
}

// 7. Fail-safe.
func TestReconcileFailSafeOnBrokenAccountManifest(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io") // minted while the manifest was good

	putReplace(t, s, "accounts/"+sqid+"/account.json", []byte("not json"))
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	// Mints nothing new for the now-unreadable account, but the existing
	// provisioned route must NOT be retired: readFailed disables removal for
	// the whole pass, even though the record has (from this pass's point of
	// view, with the account unknown) effectively vanished from the diff set.
	mustGet(t, reg, "acme.go.tito.io")

	putReplace(t, s, "accounts/"+sqid+"/account.json", accountManifestBody(acctID, sqid, "Tito"))
	if err := DeleteRecord(ctx, s, sqid, "tito", "acme.go.tito.io"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("route survived after the manifest was fixed and the record removed: %+v %v", rt, err)
	}
}

// A manifest READ FAILURE (as opposed to "no manifest") must not wipe an
// already-provisioned route's seed: seedFor's "" fallback is only safe to
// hand a brand-new mint (nothing existing to protect), never the converge
// branch. This pins the self-review fix in reconcile.go's seedFor/seedIsKnown
// split — the first passing implementation called SetSeed(host, "") here,
// which would erase a working seed on a transient blip.
//
// It also confirms readFailed from a manifest read failure disables removal
// for the WHOLE pass, not only the account-listing step the brief's step 1
// names explicitly: a second, in-scope provisioned route with no matching
// record this pass (the removal phase's only candidate) must survive too.
func TestReconcileWithholdsSeedOnManifestReadFailure(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	before := *mustGet(t, reg, "acme.go.tito.io")
	if before.Seed != `["seed"]` {
		t.Fatalf("fixture's route didn't seed as expected: %+v", before)
	}

	// A second provisioned, in-scope route with no matching record at all —
	// the removal phase's only candidate this pass.
	if err := reg.Upsert(registry.Route{Host: "ghost.go.tito.io", Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "ie", Pool: "tito", Account: sqid}); err != nil {
		t.Fatal(err)
	}

	// Break the app's instances.json: GetManifest now fails non-ENOENT.
	putReplace(t, s, ManifestKey(sqid, "tito"), []byte("not json"))

	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	after := *mustGet(t, reg, "acme.go.tito.io")
	if after.Seed != before.Seed {
		t.Errorf("manifest read failure wiped an existing route's seed: before %q after %q", before.Seed, after.Seed)
	}
	after.Seed = before.Seed
	if after != before {
		t.Errorf("manifest read failure touched more than Seed: before %+v after %+v", before, after)
	}

	mustGet(t, reg, "ghost.go.tito.io") // removal skipped for the whole pass
}

// 8. Malformed record.
func TestReconcileSkipsMalformedRecord(t *testing.T) {
	s, sqid := setup(t)
	body, err := json.Marshal(Record{Schema: 1, Host: "wrong-host.go.tito.io", Slug: "wrong",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()})
	if err != nil {
		t.Fatal(err)
	}
	// Written at the key for "acme.go.tito.io" but the body claims a
	// different host: the record disagrees with its own path.
	if _, err := s.Put(ctx, RecordKey(sqid, "tito", "acme.go.tito.io"), body, ""); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("minted from a malformed record's key host: %+v %v", rt, err)
	}
	if rt, err := reg.Get("wrong-host.go.tito.io"); err != nil || rt != nil {
		t.Errorf("minted from a malformed record's body host: %+v %v", rt, err)
	}
}

// The record is the authorization, and this pass builds a socket path and a
// database path out of it — so the host-shape invariant the console's create
// door enforces has to hold at THIS door too, not by trusting the other one.
// A record written straight into the bucket (or by a future/older writer)
// whose host is not a hostname must not become a route: nothing downstream
// re-checks it. Skipped loudly, and protected from removal for the same
// reason a corrupted body is — an unusable record is not a retired one.
func TestReconcileSkipsRecordWithInvalidHostShape(t *testing.T) {
	s, sqid := setup(t)
	const bad = "acme$evil.go.tito.io"
	body, err := json.Marshal(Record{Schema: 1, Host: bad, Slug: "acme$evil", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()})
	if err != nil {
		t.Fatal(err)
	}
	// Key and body agree — the only thing wrong here is the host itself.
	if _, err := s.Put(ctx, RecordKey(sqid, "tito", bad), body, ""); err != nil {
		t.Fatal(err)
	}
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(bad); err != nil || rt != nil {
		t.Errorf("minted a route from a record whose host is not a hostname: %+v %v", rt, err)
	}
	// The good neighbour in the same pass is untouched by the skip.
	mustGet(t, reg, "acme.go.tito.io")

	// And a route that somehow already carries that host is not treated as
	// retired: the record is there, it just cannot be used.
	if err := reg.Upsert(registry.Route{Host: bad, Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "ie", Pool: "tito", Account: sqid}); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, bad)
}

// ValidateShape is a safety net, not just the console door's policy check:
// a record can reach the bucket by paths other than the console (an older
// or future writer, or direct bucket access), and this pass must not
// materialize one whose Kind/Addr/Unit/Environment combination is
// malformed. Skipped loudly, and protected from removal for the same
// reason a corrupted body or an invalid host shape is — an unusable record
// is not a retired one.
func TestReconcileSkipsRecordWithInvalidShape(t *testing.T) {
	s, sqid := setup(t)
	// kind "service" with no addr: ValidateShape refuses this combination
	// (the platform only proxies to a service, so it must name its
	// upstream), but it agrees with its own key path and its host shape is
	// fine — the only thing wrong is the Kind/Addr combination itself.
	bad := Record{Schema: 1, Host: "bad-service.go.tito.io", Slug: "bad-service", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", Kind: "service",
		CreatedAt: time.Now()}
	body, err := json.Marshal(bad)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, RecordKey(sqid, "tito", bad.Host), body, ""); err != nil {
		t.Fatal(err)
	}
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(bad.Host); err != nil || rt != nil {
		t.Errorf("minted a route from a shape-invalid record: %+v %v", rt, err)
	}
	// The good neighbour in the same pass is untouched by the skip.
	mustGet(t, reg, "acme.go.tito.io")

	// And a route that somehow already carries that host is not treated as
	// retired: the record is there, it just cannot be used.
	if err := reg.Upsert(registry.Route{Host: bad.Host, Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "ie", Pool: "tito", Account: sqid}); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, bad.Host)
}

// ValidateShape deliberately does not inspect Channel (it validates shape
// fields, Channel is not one), but a channel-less record must not become a
// route pinned to no channel — that is issue #130's failure mode (a route
// whose channel resolves to nothing, waking with "no adoptable head")
// arriving through the MINT door instead of the converge door Task 3
// closed. Skipped loudly at the read-path gate, and protected from removal
// for the same reason a corrupted body is — an unusable record is not a
// retired one.
func TestReconcileSkipsRecordWithEmptyChannel(t *testing.T) {
	s, sqid := setup(t)
	bad := acmeRecord(sqid)
	bad.Host, bad.Slug, bad.Channel = "no-channel.go.tito.io", "no-channel", ""
	body, err := json.Marshal(bad)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(ctx, RecordKey(sqid, "tito", bad.Host), body, ""); err != nil {
		t.Fatal(err)
	}
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get(bad.Host); err != nil || rt != nil {
		t.Errorf("minted a route from a channel-less record: %+v %v", rt, err)
	}
	// The good neighbour in the same pass is untouched by the skip.
	mustGet(t, reg, "acme.go.tito.io")

	// And a route that somehow already carries that host is not treated as
	// retired: the record is there, it just cannot be used.
	if err := reg.Upsert(registry.Route{Host: bad.Host, Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "ie", Pool: "tito", Account: sqid}); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, bad.Host)
}

// CRITICAL regression (code review finding): an existing route's record
// going unparseable — not gone, just unreadable this pass — must never read
// as "the bucket retired it". Before the fix, json.Unmarshal failure and
// path-disagreement both log-and-continue without setting readFailed and
// without the host entering recs, so the removal phase deleted the route.
// A single non-additive change to Record's wire shape reaching an older
// binary would make EVERY existing record fail exactly this way, turning
// one tick into a mass fleet retirement. The fix (protect map, keyed on the
// record's KEY-derived host) is deliberately narrower than readFailed: it
// holds back only the specific corrupted host, so a genuinely retired
// neighbor is still removed in the very same pass — proven below.
func TestReconcileProtectsRouteWithCorruptedRecordBody(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io") // minted while the record was good

	// Corrupt the record body IN PLACE (a replace, not a delete) — the
	// shape a non-additive encoding change would produce on an old binary.
	putReplace(t, s, RecordKey(sqid, "tito", "acme.go.tito.io"), []byte("not json"))

	// A second, genuinely retired route: its record is deleted outright
	// (not corrupted), so it must still be removed in the SAME pass — this
	// is what proves protect is narrow, not a readFailed-style pass-wide
	// veto.
	ghost := registry.Route{Host: "ghost.go.tito.io", Kind: "instance", App: "tito",
		Status: "ready", Provisioned: true, Region: "ie", Pool: "tito", Account: sqid}
	if err := reg.Upsert(ghost); err != nil {
		t.Fatal(err)
	}

	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io") // protected: an unparseable record is not "gone"
	if rt, err := reg.Get("ghost.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a genuinely record-less route was not removed alongside a protected one: %+v %v", rt, err)
	}
}

// IMPORTANT regression (code review finding): two accounts holding a record
// for the same host must not resolve by silent last-writer-wins map
// overwrite (the losing tenant's record would vanish from the pass with no
// diagnostic). First-seen (bucket listing order) wins; the loser is skipped
// loudly and produces no second route, no crash.
func TestReconcileDuplicateHostAcrossAccountsFirstSeenWins(t *testing.T) {
	s := dirStore(t)
	sqid1 := writeAccountManifest(t, s, 2, "Tito")
	sqid2 := writeAccountManifest(t, s, 3, "Other")
	rec1 := Record{Schema: 1, Host: "dup.go.tito.io", Slug: "dup", App: "tito",
		Account: sqid1, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()}
	rec2 := Record{Schema: 1, Host: "dup.go.tito.io", Slug: "dup", App: "tito",
		Account: sqid2, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, rec1); err != nil {
		t.Fatal(err)
	}
	if err := CreateRecord(ctx, s, rec2); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	// The winner is whichever record's KEY sorts first — that is the order
	// s.List("apps/") actually hands the reconcile loop, so this must be
	// derived, never assumed.
	key1, key2 := RecordKey(sqid1, "tito", rec1.Host), RecordKey(sqid2, "tito", rec2.Host)
	winner := sqid1
	if key2 < key1 {
		winner = sqid2
	}
	got := mustGet(t, reg, "dup.go.tito.io")
	if got.Account != winner {
		t.Errorf("duplicate host resolved to account %q, want first-seen %q", got.Account, winner)
	}
	if routes, err := reg.List(); err != nil || len(routes) != 1 {
		t.Errorf("duplicate host produced %d routes, want exactly 1: %+v %v", len(routes), routes, err)
	}
}

// Trash: an app in the trash provisions nothing. Its record may still be in
// the bucket (purge is what deletes it), but minting a route for it would
// fight domains.Reconcile's trash pass — which drops a trashed app's routes
// every tick — forever. Restore lifts the refusal the very next pass.
func TestReconcileDeclinesTrashedApp(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Fatalf("minted a trashed app's instance route: %+v %v", rt, err)
	}

	if err := release.RestoreApp(ctx, s, sqid, "tito"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io")
}

// Trash: a route whose app is in the trash is RETIRED, on every box that
// serves it (#340). This is provision's job and not domains.Reconcile's,
// even though domains removes the same routes for the same reason: the
// domains pass runs only on the box holding the housekeeper lease, so on a
// multi-box deployment every other box kept a trashed app's route forever
// and went on serving it — publicly, from its own edge, with the app
// deleted. Single-box deployments were correct by accident.
//
// The record stays in the bucket while an app is trashed, so the removal
// phase's "record gone" trigger never fires for it. The trash marker is the
// second trigger, asked under exactly the same three gates: Provisioned,
// in scope, and nothing failed to read this pass.
func TestReconcileRemovesTrashedAppsRoute(t *testing.T) {
	_, pub := releaseKeys(t)
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	})
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Fatalf("a trashed app's route survived: %+v %v", rt, err)
	}
}

// #340's own shape: the FAR box, the everywhere class, a static site. This
// is the case that was live on flagship — three edges publishing a trashed
// app's host in their edges files while the lease holder had already
// dropped it, so even the lease holder served it again through the
// deployment-wide host map.
func TestReconcileRemovesTrashedEverywhereRouteOnAFarBox(t *testing.T) {
	_, pub := releaseKeys(t)
	s, sqid := setup(t)
	rec := titoRecord(sqid, "site")
	rec.Kind = "static"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := farBoxCfg()
	cfg.ReleasePub = pub
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "site.go.tito.io")

	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("site.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a far box kept serving a trashed app's static route: %+v %v", rt, err)
	}
}

// The two trash questions are NOT the same question. Declining to mint is
// reversible and answers "true" to an unreadable marker on purpose;
// removal is not reversible and must answer only to a marker positively
// read and present. A pass that conflated them would let a corrupt object
// — or a bucket writer with delete-but-not-sign — retire live routes.
func TestReconcileKeepsTrashedAppsRouteWhenTheMarkerIsUnreadable(t *testing.T) {
	_, pub := releaseKeys(t)
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	})
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	putReplace(t, s, release.AppTombstoneKey(sqid, "tito"), []byte("{not json"))
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io")
}

// The production flag is why this removal lived in domains.Reconcile at
// all: dropping every route of a production app is an outage, which is
// exactly what the flag exists to prevent. Moving the trigger without the
// guard would have been a regression, not a fix. Same refusal, same
// wording, same pass-after-pass loudness as domains §1c.
func TestReconcileNeverRemovesRoutesOfAProductionFlaggedTrashedApp(t *testing.T) {
	priv, pub := releaseKeys(t)
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	})
	// The flag goes on FIRST: SetProduction refuses a trashed app.
	if err := release.SetProduction(ctx, s, priv, sqid, "tito", true, "paul@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	if err := release.TrashApp(ctx, s, sqid, "tito", "eve@evil.example", time.Now()); err != nil {
		t.Fatal(err)
	}

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	for pass := 1; pass <= 2; pass++ {
		if err := Reconcile(ctx, s, reg, cfg); err != nil {
			t.Fatalf("pass %d: %v", pass, err)
		}
		mustGet(t, reg, "acme.go.tito.io")
	}
	if !strings.Contains(logged.String(), "trash marker on production app") {
		t.Errorf("no refusal logged; log was:\n%s", logged.String())
	}
}

// flagUnreadableStore breaks exactly one read — the app's production flag —
// the way an attacker with delete-but-not-sign, or a half-broken bucket,
// would. Everything else reads normally, so the trash marker is perfectly
// valid and only the flag is in doubt.
type flagUnreadableStore struct {
	release.Store
	key string
}

func (s flagUnreadableStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == s.key {
		return nil, "", errors.New("bucket refused the read")
	}
	return s.Store.Get(ctx, key)
}

// An UNREADABLE flag refuses in the same direction as an ON one, here as
// everywhere else in this platform: the state an attacker can manufacture
// is exactly the unreadable one.
func TestReconcileDoesNotRemoveTrashedRoutesWhenTheProductionFlagIsUnreadable(t *testing.T) {
	_, pub := releaseKeys(t)
	dir, sqid := setup(t)
	if err := CreateRecord(ctx, dir, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	})
	if err := release.TrashApp(ctx, dir, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	s := flagUnreadableStore{Store: dir, key: release.ProductionKey(sqid, "tito")}

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io")
	if !strings.Contains(logged.String(), "cannot read the production flag") {
		t.Errorf("no refusal logged; log was:\n%s", logged.String())
	}
}

// A KEYLESS box acts on the marker, deliberately — domains §1c's posture,
// carried over verbatim so the two passes cannot disagree about the same
// app. It cannot verify a flag, and the alternative — hold every trash
// forever wherever there is no pubkey — would break hiding entirely on
// legitimately keyless deployments, whose owners have no way to make the
// guard pass.
func TestReconcileRemovesTrashedAppsRouteOnAKeylessBox(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Hibernate: true, Provisioned: true, Seed: `["seed"]`, Status: "ready",
	})
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} // no ReleasePub
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Fatalf("a keyless box kept a trashed app's route: %+v %v", rt, err)
	}
}

// Provisioned is the flag that grants removal authority, and the trash
// trigger does not widen it: a hand-added route this pass never adopted is
// not this pass's to retire, whatever the app's marker says.
func TestReconcileNeverRemovesAHandAddedRouteOfATrashedApp(t *testing.T) {
	_, pub := releaseKeys(t)
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, registry.Route{
		Host: "hand.go.tito.io", Slug: "hand", Kind: "instance",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie", Channel: "stable",
		Status: "ready", // no Provisioned
	})
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "hand.go.tito.io")
}

// Trash: an unreadable or malformed marker also declines minting — not
// minting is the reversible direction, and a writer who could plant garbage
// here to stall minting can already retire the instance outright by
// deleting its record.
func TestReconcileDeclinesUnreadableTrashMarker(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	if err := release.TrashApp(ctx, s, sqid, "tito", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	putReplace(t, s, release.AppTombstoneKey(sqid, "tito"), []byte("{not json"))
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Fatalf("an unreadable trash marker still minted a route: %+v %v", rt, err)
	}
}

// Design §4.2 — the payoff. Backfilling the fleet is a pure bucket-side
// write per route: describe what is already there, and the next serial bump
// adopts it. No carlos add, no SSM session, no on-box verb.
func TestReconcileAdoptsMatchingHandAddedRoute(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	rec.Channel = "edge"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	// The route as an operator hand-added it before records existed: the
	// same shape, a different channel, and no Provisioned flag.
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}
	reg := regWith(t, handAdded)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	got := *mustGet(t, reg, "acme.go.tito.io")
	if !got.Provisioned {
		t.Fatal("a matching hand-added route was not adopted")
	}
	if got.Channel != "edge" {
		t.Errorf("channel = %q, want edge — adoption should converge in the same pass", got.Channel)
	}
}

// The match test EXCLUDES the three columns the record is taking ownership
// of. Requiring them to agree first would make adoption impossible in
// exactly the case it exists for — a hand-added route on the wrong channel
// is the whole reason to adopt it.
func TestReconcileAdoptionIgnoresOwnedColumns(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	rec.Channel = "edge"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "canary/rehearsal", Environment: "staging", Hibernate: false, Status: "ready",
	}
	reg := regWith(t, handAdded)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if !mustGet(t, reg, "acme.go.tito.io").Provisioned {
		t.Error("adoption refused over a Channel/Environment/Hibernate difference")
	}
}

// A shape that differs is left exactly as alone as it is today. This is the
// guard that stands between "adopt on match" and reconcile's removal phase
// acquiring authority over a route it does not describe.
func TestReconcileRefusesToAdoptMismatchedRoute(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	mismatched := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "other", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}
	reg := regWith(t, mismatched)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a mismatched route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a mismatched route was adopted — removal authority granted over a route the record does not describe")
	}
}

// A record has no field for Backing — registry.Route.Backing's own doc
// comment calls it a box-scoped operator column, alongside capability
// grants and tunnels — so a record can never describe a carlos-app@<host>
// unit tenant at all. Even a row that otherwise matches in every column
// must not be adopted: adoption would hand this pass's removal phase
// authority over a row it cannot even state the shape of.
func TestReconcileRefusesToAdoptBackedRoute(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
		Backing: registry.BackingUnit,
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a Backing-owned route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a carlos-app@ unit tenant was adopted — a record cannot state its Backing")
	}
}

// Tunnel is routing-decisive and OUTRANKS the Socket/Addr a record can
// state: internal/edge picks the tunnel proxy before it ever looks at
// Socket (edge.go's route-selection switch), so a tunnel-backed route can
// otherwise agree with a record on every other compared column while every
// byte it actually serves goes down a channel the record cannot see or
// reclaim (2026-08-10 review, Critical).
func TestReconcileRefusesToAdoptTunnelBoundRoute(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
		Tunnel: "pi-fleet",
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a tunnel-bound route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a tunnel-bound route was adopted — a record cannot state or reclaim its Tunnel binding")
	}
}

// Caps is the operator-granted, box-local capability-grant column, and
// registry.Delete's purge of a route's grants is already a human-gated,
// last-resort action elsewhere in this file (the shape-drift log). Adoption
// must not hand that purge to an unattended bucket-record deletion
// (2026-08-10 review, Critical).
func TestReconcileRefusesToAdoptCapsBearingRoute(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
		Caps: "port:25",
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a caps-bearing route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a caps-bearing route was adopted — its grants would now be one record deletion from purged")
	}
}

// domains.mintedRoute builds platform routes with empty Slug/Pool/Region/
// DBPath, copying Kind/Socket/Addr from the source route — so a record
// whose Slug/Pool/Region are also empty (an unconstrained-region/pool box)
// can otherwise agree with a platform-minted row on every other compared
// column. Adoption must not hand this pass's removal authority to a row
// domains.Reconcile already owns and periodically re-Upserts/removes on its
// own authority (2026-08-10 review, Important 1).
func TestReconcileRefusesToAdoptPlatformRoute(t *testing.T) {
	s, sqid := setup(t)
	rec := Record{Schema: 1, Host: "acme.go.tito.io", App: "tito", Account: sqid,
		Channel: "stable", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Status: "ready", Platform: true,
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{DataDir: "/data"} // unconstrained region, default pool only
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a platform-owned route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a platform-owned route was adopted — removal authority granted to a second pass")
	}
}

// Same guard, the other flag domains.Reconcile also uses for its own
// removal authority (2026-08-10 review, Important 1).
func TestReconcileRefusesToAdoptCanaryRoute(t *testing.T) {
	s, sqid := setup(t)
	rec := Record{Schema: 1, Host: "acme.go.tito.io", App: "tito", Account: sqid,
		Channel: "stable", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Status: "ready", Canary: true,
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a canary-owned route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a canary-owned route was adopted — removal authority granted to a second pass")
	}
}

// Nothing currently locked the derived Socket/DBPath comparison specifically
// — the only prior mismatch test differed on App, so a regression that
// dropped the DBPath comparison from shapeMismatch would leave the suite
// green (2026-08-10 review, test gap called out alongside the Important
// findings).
func TestReconcileRefusesToAdoptOnDBPathMismatch(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/other.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "acme.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after := *mustGet(t, reg, "acme.go.tito.io")
	if after != before {
		t.Errorf("a DBPath-mismatched route was touched: before %+v after %+v", before, after)
	}
	if after.Provisioned {
		t.Error("a route was adopted despite a DBPath mismatch")
	}
}

// An adopted route is removable by the pass that adopted it, on the same
// terms as one it minted: the record disappearing is the bucket's own
// decision to retire the instance.
func TestReconcileRemovesAdoptedRouteWhenRecordGoes(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: "/run/carlos/acme.go.tito.io.sock", DBPath: "/data/acme.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}
	reg := regWith(t, handAdded)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if err := DeleteRecord(ctx, s, sqid, "tito", "acme.go.tito.io"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	rt, err := reg.Get("acme.go.tito.io")
	if err != nil {
		t.Fatal(err)
	}
	if rt != nil {
		t.Errorf("adopted route survived its record's deletion: %+v", rt)
	}
}

// A dry-run pass reports what it would do and writes nothing — the backfill
// mitigation (design §4.3): a production route is adopted only after someone
// has read what adopting it would mean, because adoption is what grants the
// removal phase authority over that row. This exercises all four write
// shapes in the same pass — mint (acme.go.tito.io has no existing route),
// adopt (a hand-added route matching a record's shape but on the wrong
// channel), converge (an already-provisioned route whose channel,
// environment, hibernate and seed all disagree with its record), and remove
// (a provisioned route whose record is gone) — because a test that only
// covers minting would pass while a removal wrote through.
func TestReconcileDryRunWritesNothing(t *testing.T) {
	s, sqid := setup(t)

	// Mint: no existing route for this host.
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}

	// Adopt: a hand-added route matching the record's shape, on a
	// different channel AND environment — adoption would converge both in
	// the same pass. Two differing columns, deliberately: a dry-run report
	// that only ever mentioned channel here would leave environment's
	// change invisible to the operator reading it (design §4.3 review).
	adoptRec := Record{Schema: 1, Host: "adopt.go.tito.io", Slug: "adopt", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "edge", Environment: "staging", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, adoptRec); err != nil {
		t.Fatal(err)
	}
	adoptable := registry.Route{
		Host: "adopt.go.tito.io", Slug: "adopt", Kind: "instance",
		Socket: "/run/carlos/adopt.go.tito.io.sock", DBPath: "/data/adopt.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}

	// Converge: already provisioned (this pass or an earlier one owns it),
	// but channel, environment, hibernate and seed all disagree.
	onHibernate := true
	ownedRec := Record{Schema: 1, Host: "owned.go.tito.io", Slug: "owned", App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "edge", Environment: "staging",
		Hibernate: &onHibernate, CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, ownedRec); err != nil {
		t.Fatal(err)
	}
	owned := registry.Route{
		Host: "owned.go.tito.io", Slug: "owned", Kind: "instance",
		Socket: "/run/carlos/owned.go.tito.io.sock", DBPath: "/data/owned.go.tito.io.db",
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Provisioned: true, Status: "ready",
	}

	// Remove: provisioned, in scope, and its record is gone.
	retired := registry.Route{Host: "gone.go.tito.io", Kind: "instance", App: "tito", Account: sqid,
		Pool: "tito", Region: "ie", Status: "ready", Provisioned: true}

	reg := regWith(t, adoptable, owned, retired)
	before, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", DryRun: true}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	after, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(before, after) {
		t.Errorf("dry-run wrote to the registry:\nbefore %+v\nafter  %+v", before, after)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("dry-run minted a route: %+v %v", rt, err)
	}

	// Positive control (reviewer-flagged): the assertion above — total
	// state unchanged — would also pass if this fixture simply never
	// exercised the mint/adopt/converge/remove branches at all, e.g. if a
	// future validation change caused ownedRec to be skipped and
	// protect-ed, silently dropping the converge branch from coverage while
	// the test still reported green. Re-running the identical fixture for
	// real proves the fixture DOES drive every branch: if state came out
	// unchanged here too, the dry-run assertion above would have been
	// vacuous, not a proof of anything.
	cfg.DryRun = false
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	live, err := reg.List()
	if err != nil {
		t.Fatal(err)
	}
	if reflect.DeepEqual(before, live) {
		t.Fatal("a real pass over the identical fixture left the registry unchanged — " +
			"the dry-run assertion above proves nothing if nothing was ever going to write")
	}
}

// The other side of that coin, and the one that matters more: a route no
// record ever described is never removed, no matter how many passes run.
// Adoption is the ONLY way this pass acquires removal authority, and this
// test is what stands between a backfill and a retired production route.
func TestReconcileNeverRemovesUnadoptedRoute(t *testing.T) {
	s, sqid := setup(t)
	// A record for some OTHER host, so the pass runs a full removal phase
	// with a non-empty record set rather than bailing early.
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "console.go.tito.io", Kind: "instance", App: "tito", Account: sqid,
		Pool: "tito", Region: "ie", Channel: "stable", Addr: "127.0.0.1:8090",
		Unit: "console.service", Status: "ready",
	}
	reg := regWith(t, handAdded)
	// Upsert stamps CreatedAt; read the row back so the final comparison
	// isn't comparing against a zero value the row never actually had.
	handAdded = *mustGet(t, reg, "console.go.tito.io")
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	for i := 0; i < 3; i++ {
		if err := Reconcile(ctx, s, reg, cfg); err != nil {
			t.Fatal(err)
		}
	}
	got, err := reg.Get("console.go.tito.io")
	if err != nil {
		t.Fatal(err)
	}
	if got == nil {
		t.Fatal("a hand-added route with no record was removed")
	}
	if got.Provisioned {
		t.Error("a hand-added route with no record was adopted")
	}
	if *got != handAdded {
		t.Errorf("a hand-added route with no record was modified: %+v", got)
	}
}

// TestReconcileAdoptsStaticStoryRoute tests the backfill golden path:
// a declared static record adopts the story-shaped route that was
// hand-added before records existed. This is issue #82 backfill path.
func TestReconcileAdoptsStaticStoryRoute(t *testing.T) {
	s := dirStore(t)
	// Create a different account for story (id=3, to avoid collision with the tito account)
	storyAcctSqid := writeAccountManifest(t, s, 3, "Story")
	// Create an app manifest for story
	if err := PutManifest(ctx, s, storyAcctSqid, "story", Manifest{
		Schema: 1, App: "story", Domain: "carlosframework.com", Seed: []string{},
	}); err != nil {
		t.Fatal(err)
	}

	// Pre-Upsert the operator-added shape: a static route as it existed
	// before records were written to the bucket
	handAdded := registry.Route{
		Host: "story.carlosframework.com", Kind: "static",
		App: "story", Account: storyAcctSqid, Channel: "canary/live",
		Status: "ready",
	}
	reg := regWith(t, handAdded)

	// Write the matching declared record
	rec := Record{
		Schema: 1, Host: "story.carlosframework.com", App: "story",
		Account: storyAcctSqid, Channel: "canary/live", Kind: "static",
		CreatedAt: time.Now(),
	}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}

	// Run Reconcile with an unconstrained region/pool (matches the static route)
	cfg := ReconcileConfig{DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	// Assert the route is now Provisioned
	got := *mustGet(t, reg, "story.carlosframework.com")
	if !got.Provisioned {
		t.Fatal("the matching static route was not adopted")
	}

	// Static routes should have no Socket or DBPath
	if got.Socket != "" || got.DBPath != "" {
		t.Errorf("a static route derived filesystem paths: socket %q db %q", got.Socket, got.DBPath)
	}

	// Static routes should not be hibernating
	if got.Hibernate {
		t.Error("a static route was armed for hibernation — it has no process to wake")
	}
}

// Companion negative: a static record does NOT adopt an instance route.
// The kind must match for adoption to happen.
func TestReconcileRefusesAdoptInstanceRouteForStaticRecord(t *testing.T) {
	s := dirStore(t)
	// Create a different account for story (id=3)
	storyAcctSqid := writeAccountManifest(t, s, 3, "Story")
	// Create an app manifest for story
	if err := PutManifest(ctx, s, storyAcctSqid, "story", Manifest{
		Schema: 1, App: "story", Domain: "carlosframework.com", Seed: []string{},
	}); err != nil {
		t.Fatal(err)
	}

	// Pre-Upsert a route with Kind: "instance" (not "static")
	handAdded := registry.Route{
		Host: "story.carlosframework.com", Kind: "instance",
		Socket: "/run/carlos/story.carlosframework.com.sock",
		DBPath: "/data/story.carlosframework.com.db",
		App:    "story", Account: storyAcctSqid, Channel: "canary/live",
		Status: "ready",
	}
	reg := regWith(t, handAdded)
	before := *mustGet(t, reg, "story.carlosframework.com")

	// Write the matching declared record WITH Kind: "static"
	rec := Record{
		Schema: 1, Host: "story.carlosframework.com", App: "story",
		Account: storyAcctSqid, Channel: "canary/live", Kind: "static",
		CreatedAt: time.Now(),
	}
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}

	// Run Reconcile
	cfg := ReconcileConfig{DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}

	// Assert the route was NOT adopted (Provisioned still false)
	after := *mustGet(t, reg, "story.carlosframework.com")
	if after.Provisioned {
		t.Error("a kind-mismatched route was adopted — removal authority granted over a route the record does not describe")
	}

	// Verify the route was not modified at all
	if after != before {
		t.Errorf("a mismatched route was modified: before %+v after %+v", before, after)
	}
}

// ---- serving classes (multi-region ingress design, Part 3's table) ---------
//
// The same deployment seen from two PLATFORM boxes: the record's home region
// ("ie") and another one ("us"), both serving the tito pool. Everything below
// asks the same question of one record — "does the far box materialize it?" —
// and the answer is the record's serving class.

func homeBoxCfg() ReconcileConfig {
	return ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
}

func farBoxCfg() ReconcileConfig {
	return ReconcileConfig{Region: "us", Pools: []string{"tito"}, DataDir: "/data"}
}

// titoRecord is the tito-app fixture every serving-class case below mutates:
// homed in "ie", in the "tito" pool, on the stable channel.
func titoRecord(sqid, host string) Record {
	return Record{Schema: 1, Host: host + ".go.tito.io", Slug: host, App: "tito",
		Account: sqid, Pool: "tito", Region: "ie", Channel: "stable", CreatedAt: time.Now()}
}

// A static (bake-served) record is materialized by EVERY platform box in the
// deployment, not only its home: the edge resolves the app's channel pointer
// out of the same bucket every box already reads, so no box needs the home box
// to serve it. Spec Part 3, class "Everywhere".
func TestReconcileStaticIsMaterializedOnEveryPlatformBox(t *testing.T) {
	s, sqid := setup(t)
	rec := titoRecord(sqid, "site")
	rec.Kind = "static"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, farBoxCfg()); err != nil {
		t.Fatal(err)
	}
	got := mustGet(t, reg, "site.go.tito.io")
	if !got.Provisioned || got.Kind != "static" {
		t.Fatalf("far box did not materialize the static record: %+v", *got)
	}
	// The row still carries the RECORD's region — the record is the desired
	// state and routeFor is its one renderer, so a far box's copy and the home
	// box's copy are the same route, byte for byte.
	if got.Region != "ie" {
		t.Errorf("Region = %q, want the record's own %q", got.Region, "ie")
	}
	if got.Socket != "" || got.DBPath != "" || got.Hibernate {
		t.Errorf("a static route grew a process shape on the far box: %+v", *got)
	}
}

// An addr route whose upstreams are all publicly dialable is materialized
// everywhere too — every edge probes and dials the same public backends. Spec
// Part 3, class "Everywhere, probed".
func TestReconcilePublicAddrIsMaterializedOnEveryPlatformBox(t *testing.T) {
	s, sqid := setup(t)
	one := titoRecord(sqid, "pub")
	one.Kind = "service"
	one.Addr = "203.0.113.10:443"
	many := titoRecord(sqid, "pubs")
	many.Kind = "service"
	many.Addrs = []string{"203.0.113.11:443", "198.51.100.7:443"}
	for _, rec := range []Record{one, many} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, farBoxCfg()); err != nil {
		t.Fatal(err)
	}
	for _, host := range []string{"pub.go.tito.io", "pubs.go.tito.io"} {
		if got := mustGet(t, reg, host); !got.Provisioned {
			t.Errorf("%s: far box did not materialize a public-upstream record: %+v", host, *got)
		}
	}
}

// The home-boxed classes, unchanged: the process backing them exists on
// exactly ONE box, so a far box must still refuse every one of them. This is
// the refusal test the class change is only safe with.
func TestReconcileHomeBoxedClassesStayHomeBoxed(t *testing.T) {
	s, sqid := setup(t)
	socketed := titoRecord(sqid, "exec") // no kind, no addr: the socket/exec shape
	loopback := titoRecord(sqid, "loop")
	loopback.Kind = "service"
	loopback.Addr = "127.0.0.1:8090"
	private := titoRecord(sqid, "priv")
	private.Kind = "service"
	private.Addrs = []string{"10.0.0.5:3000"}
	unitOwned := titoRecord(sqid, "unit")
	unitOwned.Addr = "203.0.113.20:8090"
	unitOwned.Unit = "console.service"
	named := titoRecord(sqid, "named")
	named.Kind = "service"
	named.Addr = "backend.example.com:443"
	for _, rec := range []Record{socketed, loopback, private, unitOwned, named} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, farBoxCfg()); err != nil {
		t.Fatal(err)
	}
	for _, host := range []string{"exec.go.tito.io", "loop.go.tito.io", "priv.go.tito.io",
		"unit.go.tito.io", "named.go.tito.io"} {
		if rt, err := reg.Get(host); err != nil || rt != nil {
			t.Errorf("%s: a home-boxed record materialized on a foreign box: %+v %v", host, rt, err)
		}
	}
	// ...and every one of them still materializes on its OWN box.
	if err := Reconcile(ctx, s, reg, homeBoxCfg()); err != nil {
		t.Fatal(err)
	}
	for _, host := range []string{"exec.go.tito.io", "loop.go.tito.io", "priv.go.tito.io",
		"unit.go.tito.io", "named.go.tito.io"} {
		mustGet(t, reg, host)
	}
}

// Pool is a TENANCY boundary, not a geography: widening the region half of the
// gate must not widen the pool half. A static record in the tito pool stays
// off a box that does not serve that pool, however far it travels.
func TestReconcileEverywhereClassStillRespectsThePool(t *testing.T) {
	s, sqid := setup(t)
	rec := titoRecord(sqid, "site")
	rec.Kind = "static"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	// A far box serving only the shared "" pool (nil Pools), and one serving
	// somebody else's pool.
	for _, cfg := range []ReconcileConfig{
		{Region: "us", DataDir: "/data"},
		{Region: "us", Pools: []string{"eleven"}, DataDir: "/data"},
	} {
		if err := Reconcile(ctx, s, reg, cfg); err != nil {
			t.Fatal(err)
		}
		if rt, err := reg.Get("site.go.tito.io"); err != nil || rt != nil {
			t.Fatalf("pools %v: a static record crossed a pool boundary: %+v %v", cfg.Pools, rt, err)
		}
	}
}

// Fleet placement is the other tenancy boundary, and it is refused from BOTH
// sides: a customer fleet's static site is not served by the platform's boxes,
// and a customer's fleet box does not pick up the platform's static records.
// (registry.FleetPlacement is the grammar — "fleet/<name>" in the Region.)
func TestReconcileFleetPlacementIsNeverEverywhere(t *testing.T) {
	s, sqid := setup(t)
	onFleet := titoRecord(sqid, "pisite")
	onFleet.Kind = "static"
	onFleet.Region = "fleet/pi"
	platform := titoRecord(sqid, "site")
	platform.Kind = "static"
	for _, rec := range []Record{onFleet, platform} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)

	// A platform box must not serve the fleet's site...
	if err := Reconcile(ctx, s, reg, farBoxCfg()); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("pisite.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a fleet-placed static record materialized on a platform box: %+v %v", rt, err)
	}
	// ...and the fleet's own box must not pick up the platform's.
	fleetReg := regWith(t)
	if err := Reconcile(ctx, s, fleetReg, ReconcileConfig{Region: "fleet/pi", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, err := fleetReg.Get("site.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a platform static record materialized on a customer fleet box: %+v %v", rt, err)
	}
	mustGet(t, fleetReg, "pisite.go.tito.io") // its own placement, unchanged
}

// The other half of adopting everywhere: retiring everywhere. A far box that
// mints an everywhere-class route from a record must remove it when that
// record goes, or the widened mint gate leaves an immortal route serving a
// retired site on every box but its home.
func TestReconcileRemovesEverywhereRouteWhenItsRecordGoes(t *testing.T) {
	s, sqid := setup(t)
	rec := titoRecord(sqid, "site")
	rec.Kind = "static"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := farBoxCfg()
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "site.go.tito.io")

	if err := DeleteRecord(ctx, s, sqid, "tito", "site.go.tito.io"); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("site.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a far box's everywhere-class route outlived its record: %+v %v", rt, err)
	}
}

// TestEverywhereClassAgreesAcrossRecordAndRow is the drift pin between the two
// halves of the class rule: everywhereClass reads a RECORD (the mint gate) and
// everywhereRoute reads a ROW (the removal gate, asked when the record is gone
// and the row is all that is left). routeFor is the one renderer between them,
// so the property holds by construction today — and it is exactly what a class
// rule added to one function alone would break, silently, with every other test
// in this file still green. That divergence is the immortality bug the pair was
// written to close: a far box that mints an everywhere-class route and then
// cannot see it as one serves a retired site forever.
//
// The fixture set is every shape the file already exercises, plus the two
// directions a new rule would most plausibly be written into one side only (a
// probe config, an explicitly non-hibernating addr record).
func TestEverywhereClassAgreesAcrossRecordAndRow(t *testing.T) {
	off := false
	on := true
	base := func(host string) Record {
		return Record{Schema: 1, Host: host, Slug: "s", App: "tito", Account: "bac",
			Pool: "tito", Region: "ie", Channel: "stable"}
	}
	with := func(host string, mut func(*Record)) Record {
		rec := base(host)
		mut(&rec)
		return rec
	}
	fixtures := []struct {
		name string
		rec  Record
	}{
		{"socket instance (no kind, no addr)", base("exec.go.tito.io")},
		{"static", with("site.go.tito.io", func(r *Record) { r.Kind = "static" })},
		{"static, status-named", with("named-site.go.tito.io", func(r *Record) { r.Kind, r.StatusNamed = "static", &on })},
		{"public single addr", with("pub.go.tito.io", func(r *Record) { r.Kind, r.Addr = "service", "203.0.113.10:443" })},
		{"public addrs set", with("pubs.go.tito.io", func(r *Record) {
			r.Kind, r.Addrs = "service", []string{"203.0.113.11:443", "198.51.100.7:443"}
		})},
		{"public addrs with an explicit probe", with("probed.go.tito.io", func(r *Record) {
			r.Kind, r.Addrs, r.Health = "service", []string{"203.0.113.12:443", "198.51.100.8:443"}, &Health{Path: "/healthz"}
		})},
		{"public addr, explicitly never hibernating", with("awake.go.tito.io", func(r *Record) {
			r.Kind, r.Addr, r.Hibernate = "service", "203.0.113.13:443", &off
		})},
		{"loopback addr", with("loop.go.tito.io", func(r *Record) { r.Kind, r.Addr = "service", "127.0.0.1:8090" })},
		{"private addrs", with("priv.go.tito.io", func(r *Record) { r.Kind, r.Addrs = "service", []string{"10.0.0.5:3000"} })},
		{"mixed public and private addrs", with("mixed.go.tito.io", func(r *Record) {
			r.Kind, r.Addrs = "service", []string{"203.0.113.14:443", "192.168.1.9:3000"}
		})},
		{"name-based addr", with("named.go.tito.io", func(r *Record) { r.Kind, r.Addr = "service", "backend.example.com:443" })},
		{"unit-owned public addr", with("unit.go.tito.io", func(r *Record) {
			r.Addr, r.Unit = "203.0.113.20:8090", "console.service"
		})},
		{"environment-bound static", with("env-site.go.tito.io", func(r *Record) { r.Kind, r.Environment = "static", "staging" })},
		{"fleet-placed static", with("pisite.go.tito.io", func(r *Record) { r.Kind, r.Region = "static", "fleet/pi" })},
	}
	for _, f := range fixtures {
		t.Run(f.name, func(t *testing.T) {
			// Every fixture must be one the pass would actually materialize —
			// a shape ValidateShape refuses never reaches either gate, so a
			// property proved on one would prove nothing.
			if err := ValidateShape(f.rec); err != nil {
				t.Fatalf("fixture is not a shape this pass would ever see: %v", err)
			}
			fromRecord := everywhereClass(f.rec)
			fromRow := everywhereRoute(RouteFor(f.rec, "/data", `["seed"]`, false, nil))
			if fromRecord != fromRow {
				t.Fatalf("class disagrees across the mint and removal gates: everywhereClass = %v, "+
					"everywhereRoute(RouteFor(...)) = %v — a record this box mints must be a row it can retire",
					fromRecord, fromRow)
			}
		})
	}
}

// TestRouteForCarriesAddrsAndHealth is Task 3's mint-side proof: a record
// with an explicit multi-upstream Addrs set mints a route whose Addrs is the
// comma-joined string and whose Health carries a fully-materialized default
// probe (design §Part II). A legacy single-Addr record mints a
// byte-identical route to before this field existed — routeFor only ever
// reads the record's EXPLICIT Addrs, never AddrsOrDefault's Addr-folding.
func TestRouteForCarriesAddrsAndHealth(t *testing.T) {
	rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable",
		Kind: "service", Addrs: []string{"10.0.0.1:3000", "10.0.0.2:3000"}}
	rt := RouteFor(rec, "/data", "seed", false, nil)
	if rt.Addrs != "10.0.0.1:3000,10.0.0.2:3000" {
		t.Fatalf("Addrs = %q", rt.Addrs)
	}
	if rt.Health == "" {
		t.Fatal("multi-addr record must mint default health JSON")
	}
	var h Health
	if err := json.Unmarshal([]byte(rt.Health), &h); err != nil || h.Path != "/" || h.FailAfter != 3 {
		t.Fatalf("health JSON = %q (%v)", rt.Health, err)
	}
	// Legacy single-addr record: byte-identical route to before this change.
	old := Record{Host: "s.example.com", App: "web", Account: "bab", Channel: "stable", Kind: "service", Addr: "10.0.0.9:3000"}
	ort := RouteFor(old, "/data", "seed", false, nil)
	if ort.Addrs != "" || ort.Health != "" || ort.Addr != "10.0.0.9:3000" {
		t.Fatalf("legacy record's route changed shape: %+v", ort)
	}
}

// TestRouteForMaterializesPartialHealthDefaults is the controller amendment
// from Task 1's review: a record with EXPLICIT-but-partial probe config
// (only Path set here) must still mint a route whose Health JSON carries the
// platform's fully-materialized defaults for every field the record left at
// zero. An un-defaulted interval_ms of 0 reaching the edge would mean a
// tight-loop probe against a customer upstream.
func TestRouteForMaterializesPartialHealthDefaults(t *testing.T) {
	rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable",
		Kind: "service", Addrs: []string{"10.0.0.1:3000", "10.0.0.2:3000"},
		Health: &Health{Path: "/healthz"}}
	rt := RouteFor(rec, "/data", "seed", false, nil)
	var h Health
	if err := json.Unmarshal([]byte(rt.Health), &h); err != nil {
		t.Fatalf("health JSON = %q (%v)", rt.Health, err)
	}
	if h.Path != "/healthz" {
		t.Fatalf("explicit Path lost: %+v", h)
	}
	if h.IntervalMS != 2000 {
		t.Fatalf("partial probe config minted an un-defaulted interval: %+v", h)
	}
	if h.FailAfter != 3 {
		t.Fatalf("partial probe config minted an un-defaulted fail_after: %+v", h)
	}
}

// TestShapeMismatchComparesAddrsAndHealth is Task 3's adopt-side proof:
// shapeMismatch must refuse to adopt a hand-added route whose Addrs or
// Health disagree with what the record describes, the same guard it already
// applies to the legacy Addr column.
func TestShapeMismatchComparesAddrsAndHealth(t *testing.T) {
	a := registry.Route{Kind: "service", Addrs: "x:1,y:2"}
	b := registry.Route{Kind: "service", Addrs: "x:1,z:3"}
	if shapeMismatch(a, b) == "" {
		t.Fatal("differing addrs adopted silently")
	}
	c := registry.Route{Kind: "service", Addrs: "x:1,y:2", Health: `{"path":"/up"}`}
	if shapeMismatch(a, c) == "" {
		t.Fatal("differing health adopted silently")
	}
	if shapeMismatch(a, registry.Route{Kind: "service", Addrs: "x:1,y:2"}) != "" {
		t.Fatal("equal shapes must match")
	}
}

// --- inbound placement (design §8, box half) --------------------------------

// TestReconcileNoInboundBoxRefusesRoutedInstances pins the instance pass's
// half of design §8: a box that has declared CARLOS_INBOUND=none must never
// mint a routed instance row, no matter what an account's own record asks
// for — a box that cannot serve inbound traffic must never be handed
// something to serve. This is independent of ReconcileSidecars' placement
// gate (Task 9): that gate matches a RECORD's placement request against the
// box; this one refuses ALL ordinary instance minting outright, because an
// ordinary instance record carries no placement vocabulary at all.
func TestReconcileNoInboundBoxRefusesRoutedInstances(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", Inbound: InboundNone}

	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Fatalf("a no-inbound box minted a routed instance: %+v %v", rt, err)
	}
	if !strings.Contains(logged.String(), "acme.go.tito.io") || !strings.Contains(logged.String(), "no-inbound") {
		t.Fatalf("refusal must be logged loudly (host + reason), got: %s", logged.String())
	}
}

// TestNormalizeInboundRestrictiveVocabulary pins the case/whitespace-folded
// "none" spellings (review fix, Task 10 finding): "none", "NONE" and
// "  None  " must all normalize to the one restrictive value the mint/
// placement gates compare against with exact equality, regardless of how an
// operator typed it.
func TestNormalizeInboundRestrictiveVocabulary(t *testing.T) {
	for _, raw := range []string{"none", "NONE", "  None  "} {
		if got := NormalizeInbound(raw); got != InboundNone {
			t.Errorf("NormalizeInbound(%q) = %q, want %q", raw, got, InboundNone)
		}
	}
}

// TestNormalizeInboundInertVocabulary pins the case/whitespace-folded
// "any"/unset spellings: "", "any" and "ANY" must all normalize to a value
// that is NOT InboundNone, and a box configured with any of them must mint a
// routed instance exactly as it always has — the closed vocabulary must
// never make a deliberate "any" spelling behave like an unrecognised one.
func TestNormalizeInboundInertVocabulary(t *testing.T) {
	for _, raw := range []string{"", "any", "ANY"} {
		got := NormalizeInbound(raw)
		if got == InboundNone {
			t.Fatalf("NormalizeInbound(%q) = %q, must not be restrictive", raw, got)
		}

		s, sqid := setup(t)
		if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
			t.Fatal(err)
		}
		reg := regWith(t)
		cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", Inbound: got}
		if err := Reconcile(ctx, s, reg, cfg); err != nil {
			t.Fatal(err)
		}
		mustGet(t, reg, "acme.go.tito.io")
	}
}

// TestNormalizeInboundUnrecognisedIsRestrictiveAndLogged pins the review
// fix's ruling directly (Task 10 finding): the only way to produce an
// unrecognised CARLOS_INBOUND value is to have typed one, and an operator
// only sets this variable at all when they mean to restrict — so an
// unrecognised value must normalize to the restrictive reading AND be
// logged loudly, naming the offending value, so a fat-fingered box is
// caught on the next log tail rather than by a misplaced workload.
func TestNormalizeInboundUnrecognisedIsRestrictiveAndLogged(t *testing.T) {
	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	got := NormalizeInbound("off")
	if got != InboundNone {
		t.Fatalf("NormalizeInbound(%q) = %q, want restrictive %q", "off", got, InboundNone)
	}
	if !strings.Contains(logged.String(), "off") {
		t.Fatalf("unrecognised value must be logged by name, got: %s", logged.String())
	}
}

// TestReconcileUnsetInboundMintsAsBefore pins the inert-default half: a box
// whose CARLOS_INBOUND is unset (the zero value read by provisionTickConfig)
// must mint exactly as it always has — the new guard must never fire for
// "any"/"" boxes, which is every box that exists today.
func TestReconcileUnsetInboundMintsAsBefore(t *testing.T) {
	s, sqid := setup(t)
	if err := CreateRecord(ctx, s, acmeRecord(sqid)); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	mustGet(t, reg, "acme.go.tito.io")
}

// TestRouteForCarriesSteering: the mint renders the record's steering opt-in
// onto the route, byte for byte — no derivation, unlike Hibernate/StatusNamed.
func TestRouteForCarriesSteering(t *testing.T) {
	rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable", Steering: "latency"}
	if got := RouteFor(rec, "/data", "seed", false, nil).Steering; got != "latency" {
		t.Fatalf("Steering = %q, want latency", got)
	}
	unsteered := Record{Host: "s.example.com", App: "web", Account: "bab", Channel: "stable"}
	if got := RouteFor(unsteered, "/data", "seed", false, nil).Steering; got != "" {
		t.Fatalf("Steering = %q, want empty for a record that never mentioned it", got)
	}
}

// TestShapeMismatchIgnoresSteering is the re-ruled adopt-side proof (plan
// amendment b485ec5, review finding 2): Steering is RECORD-OWNED, the same
// category Channel/Environment/Hibernate are — a hand-added route whose
// steering disagrees with what the record describes must still be ADOPTED,
// not refused, because comparing it here would permanently block adoption
// for exactly the case adoption exists to serve (an operator's hand-added
// route predating the console's steering PATCH door). convergeOwned stamps
// the record's Steering onto the row unconditionally once adoption succeeds
// — this test's first assertion is what proves that stamp isn't blocked.
func TestShapeMismatchIgnoresSteering(t *testing.T) {
	a := registry.Route{Kind: "instance", Steering: "latency"}
	b := registry.Route{Kind: "instance", Steering: ""}
	if shapeMismatch(a, b) != "" {
		t.Fatalf("differing steering blocked adoption: %q, want a match — steering is record-owned", shapeMismatch(a, b))
	}
	if shapeMismatch(a, registry.Route{Kind: "instance", Steering: "latency"}) != "" {
		t.Fatal("equal steering must match")
	}
	// A steering-only difference must not mask a REAL mismatch on another
	// column — shapeMismatch still has to walk past it to the next case.
	c := registry.Route{Kind: "service", Steering: "latency"}
	if reason := shapeMismatch(a, c); reason == "" {
		t.Fatal("a genuine kind mismatch was hidden behind the ignored steering column")
	}
}

// --- #162: replica records, the home's placement set, and the two boxes' --
// --- division of labour ---------------------------------------------------

// TestReplicaRecordMintsOnlyOnItsPlacement: a replica record names ONE
// placement — its Region — and only the box carrying that placement mints
// it. Everywhere else the record is out of scope exactly like any other
// region's record: a replica minted on the wrong region would be a process
// holding live order data nobody authorized into that region (the residence
// rule), and a mint on the HOME box would stand up a second process of the
// same account on a box that already holds the authoritative copy.
func TestReplicaRecordMintsOnlyOnItsPlacement(t *testing.T) {
	s, sqid := setup(t)
	home := acmeRecord(sqid)
	if err := CreateRecord(ctx, s, home); err != nil {
		t.Fatal(err)
	}
	replica := Record{Schema: 1, Host: "acme-uk.go.tito.io", Slug: "acme-uk", App: "tito",
		Account: sqid, Pool: "tito", Region: "uk", Channel: "stable",
		Role: "replica", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, replica); err != nil {
		t.Fatal(err)
	}

	// The uk box mints the replica.
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	got, err := reg.Get("acme-uk.go.tito.io")
	if err != nil {
		t.Fatal(err)
	}
	if got == nil {
		t.Fatal("the uk box did not mint the replica record homed in uk")
	}

	// The ie box (the home's region) mints the home and nothing else.
	regIE := regWith(t)
	if err := Reconcile(ctx, s, regIE, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, _ := regIE.Get("acme-uk.go.tito.io"); rt != nil {
		t.Fatal("the ie box minted a record whose replica placement is uk — out of scope")
	}
	if rt, _ := regIE.Get("acme.go.tito.io"); rt == nil {
		t.Fatal("the ie box did not mint the home record")
	}
}

// TestHomeRecordConvergesReplicaSet: the home record's placement set is
// authoritative in BOTH directions — the one deliberate widening the owned
// arm permits — because a set that shrank must actually shrink on the row,
// or the edges file goes on publishing a replica placement the account no
// longer has. That is a residence lie, not a cosmetic one.
func TestHomeRecordConvergesReplicaSet(t *testing.T) {
	s, sqid := setup(t)
	home := acmeRecord(sqid)
	home.Replicas = []string{"uk", "ie"}
	if err := CreateRecord(ctx, s, home); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}

	if rt, _ := reg.Get("acme.go.tito.io"); rt == nil {
		t.Fatal("home not minted")
	} else if rt.Replicas != "ie,uk" {
		t.Fatalf("minted row replicas = %q, want the canonical sorted set %q", rt.Replicas, "ie,uk")
	}

	// The set shrinks in the bucket; the row follows on the next pass.
	home.Replicas = []string{"uk"}
	if _, err := updateRecord(ctx, s, sqid, "tito", "acme.go.tito.io", func(r *Record) error { r.Replicas = []string{"uk"}; return nil }); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get("acme.go.tito.io"); rt == nil {
		t.Fatal("home vanished")
	} else if rt.Replicas != "uk" {
		t.Fatalf("row replicas = %q after the record's set shrank, want %q — a stale placement keeps a retired region serving the account", rt.Replicas, "uk")
	}

	// And to none: the mirror deleted its last replica.
	if _, err := updateRecord(ctx, s, sqid, "tito", "acme.go.tito.io", func(r *Record) error { r.Replicas = nil; return nil }); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get("acme.go.tito.io"); rt == nil {
		t.Fatal("home vanished")
	} else if rt.Replicas != "" {
		t.Fatalf("row replicas = %q after the record cleared its set, want empty", rt.Replicas)
	}
}

// TestReplicaMarkMintsAndConverges: the row's replica mark is Record.Role's
// shadow, and the mark decides how the edges file publishes the row — so it
// converges from the record in BOTH directions. A re-mint carrying no role
// must DEMOTE a row that carried one: a stale mark keeps steering treating
// the row's box as a replica of an account that no longer has one there.
func TestReplicaMarkMintsAndConverges(t *testing.T) {
	s, sqid := setup(t)
	replica := Record{Schema: 1, Host: "acme-uk.go.tito.io", Slug: "acme-uk", App: "tito",
		Account: sqid, Pool: "tito", Region: "uk", Channel: "stable",
		Role: "replica", CreatedAt: time.Now()}
	if err := CreateRecord(ctx, s, replica); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get("acme-uk.go.tito.io"); rt == nil {
		t.Fatal("replica not minted")
	} else if !rt.Replica {
		t.Fatal("minted row carries no replica mark")
	}

	// The role leaves the record; the mark follows.
	if _, err := updateRecord(ctx, s, sqid, "tito", "acme-uk.go.tito.io", func(r *Record) error { r.Role = ""; return nil }); err != nil {
		t.Fatal(err)
	}
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	rt, err := reg.Get("acme-uk.go.tito.io")
	if err != nil || rt == nil {
		t.Fatalf("get: %v", err)
	}
	if rt.Replica {
		t.Fatal("row keeps a replica mark its record no longer carries — the edges file would go on telling the fleet this box replicas an account it homes")
	}
}

// TestReplicaNeverEverywhereClass pins the residence rule's second door
// (Fable review, finding 3): a replica is a process on ONE box — the one its
// placement names — and everywhereClass's widening is the mint gate for
// every non-fleet box. The class check refuses a replica FIRST (before any
// kind or addr test), and everywhereRoute agrees, because the two must
// never disagree about a row whose record is gone.
func TestReplicaNeverEverywhereClass(t *testing.T) {
	// Static would pass the kind arm on its own — the role check must be
	// first and absolute.
	if everywhereClass(Record{Role: "replica", Kind: "static"}) {
		t.Error("everywhereClass(replica static) = true — a replica must never widen past its placement box")
	}
	if everywhereClass(Record{Role: "replica", Kind: "instance"}) {
		t.Error("everywhereClass(replica instance) = true")
	}
	// A public-addr replica would pass the dial test on its own; the shape
	// door refuses it, and the class check must not depend on that door
	// holding.
	if err := ValidateShape(Record{Role: "replica", Addr: "203.0.113.5:8090"}); err == nil {
		t.Error("ValidateShape(replica with a public addr) accepted — the everywhere-class second door")
	}
	if everywhereRoute(registry.Route{Replica: true, Kind: "static"}) {
		t.Error("everywhereRoute(replica row) = true — the removal-phase half of the same question")
	}
}

// A fresh row on a box whose layout intent is nested is minted at the
// nested paths (design §8.6 step 5, plan 5), never flat and then moved. A
// flat or absent intent mints the flat defaults as before; an unflagged
// (externally supervised) socket route stays flat whatever the intent; an
// intent that cannot be read defers the mint rather than guessing.
func TestReconcileMintsNestedUnderANestedIntent(t *testing.T) {
	s, sqid := setup(t)
	off := false
	recs := []Record{
		acmeRecord(sqid),
		{Schema: 1, Host: "router.go.tito.io", Slug: "router", App: "tito", Account: sqid,
			Pool: "tito", Region: "ie", Channel: "stable", Hibernate: &off, CreatedAt: time.Now()},
	}
	for _, rec := range recs {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	reg := regWith(t)
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data",
		Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	got := *mustGet(t, reg, "acme.go.tito.io")
	if got.DBPath != NestedDBPath("/data", "acme.go.tito.io") || got.Socket != NestedSocket("acme.go.tito.io") || !got.Hibernate {
		t.Errorf("minted under a nested intent = db %q socket %q hibernate %v, want the nested paths", got.DBPath, got.Socket, got.Hibernate)
	}
	ext := *mustGet(t, reg, "router.go.tito.io")
	if ext.DBPath != DefaultDBPath("/data", "router.go.tito.io") || ext.Socket != DefaultSocket("router.go.tito.io") {
		t.Errorf("an unflagged socket route minted nested: db %q socket %q", ext.DBPath, ext.Socket)
	}
	// The next pass sees the nested row as the record's own, not drift,
	// and converges nothing on it.
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if again := *mustGet(t, reg, "acme.go.tito.io"); again.DBPath != got.DBPath || again.Socket != got.Socket {
		t.Errorf("second pass moved the row: %+v", again)
	}

	// Flat intent, and no intent: the flat defaults.
	for name, layout := range map[string]func(context.Context, release.Store) (string, error){
		"flat": func(context.Context, release.Store) (string, error) { return release.LayoutModeFlat, nil },
		"none": func(context.Context, release.Store) (string, error) { return release.LayoutModeNone, nil },
		"nil":  nil,
	} {
		reg := regWith(t)
		if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", Layout: layout}); err != nil {
			t.Fatal(err)
		}
		if got := *mustGet(t, reg, "acme.go.tito.io"); got.DBPath != "/data/acme.go.tito.io.db" {
			t.Errorf("%s intent: db %q, want the flat default", name, got.DBPath)
		}
	}

	// Unreadable: nothing is minted this pass, and the pass itself is not
	// an error (the records are retried next pass).
	reg = regWith(t)
	calls := 0
	cfg.Layout = func(context.Context, release.Store) (string, error) { calls++; return "", errors.New("store down") }
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := reg.Get("acme.go.tito.io"); err != nil || rt != nil {
		t.Errorf("a row was minted while the intent was unreadable: %+v %v", rt, err)
	}
	if calls != 1 {
		t.Errorf("the intent was read %d times in one pass, want once", calls)
	}
}

// A hand-added row at the nested paths (`carlos add` on a nested box) is
// the record's route and is adopted, exactly as a flat hand-added one is.
func TestReconcileAdoptsAHandAddedNestedRoute(t *testing.T) {
	s, sqid := setup(t)
	rec := acmeRecord(sqid)
	rec.Channel = "edge"
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	handAdded := registry.Route{
		Host: "acme.go.tito.io", Slug: "acme", Kind: "instance",
		Socket: NestedSocket("acme.go.tito.io"), DBPath: NestedDBPath("/data", "acme.go.tito.io"),
		App: "tito", Account: sqid, Pool: "tito", Region: "ie",
		Channel: "stable", Hibernate: true, Status: "ready",
	}
	reg := regWith(t, handAdded)
	if err := Reconcile(ctx, s, reg, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
		t.Fatal(err)
	}
	got := *mustGet(t, reg, "acme.go.tito.io")
	if !got.Provisioned || got.Channel != "edge" {
		t.Fatalf("a nested hand-added route was not adopted: %+v", got)
	}
	if got.DBPath != handAdded.DBPath || got.Socket != handAdded.Socket {
		t.Errorf("adoption rewrote the nested paths: %+v", got)
	}
}

// ---- move holds (moves spec §5.2, §9.6, §19 K9) ----

// movingRecord is acme with a Phase 1 move from ie to uk after the flip: the
// record's region is already the target's.
func movingRecord(sqid string) Record {
	rec := acmeRecord(sqid)
	rec.Region = "uk"
	rec.Move = &MoveState{ID: "mv_acme", From: "ie", To: "uk", State: MoveLanding, LandingPhase: LandChecking,
		RequestedAt: time.Now(), UpdatedAt: time.Now()}
	return rec
}

// holdLog records the hold writes a pass reports through HoldWritten.
type holdLog struct{ hosts []string }

func (h *holdLog) written(host string) { h.hosts = append(h.hosts, host) }

// Review Focus 5: on the target the landing row is minted held and never
// ready, at flat paths whatever the box's layout intent says (G11); on the
// source, a participant's moving row survives the record naming another
// region, and without the participant answer it is removed as today.
func TestReconcileHoldsParticipantRows(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	hold := func(r Record) (string, string, bool) {
		if r.Host == rec.Host && r.Move != nil {
			return "landing", r.Move.ID, true
		}
		return "", "", false
	}

	// The uk box: minted held, twice over, never ready; flat paths.
	uk := regWith(t)
	var wrote holdLog
	ukCfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", Hold: hold, HoldWritten: wrote.written,
		Layout: func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }}
	for pass := 1; pass <= 2; pass++ {
		if err := Reconcile(ctx, s, uk, ukCfg); err != nil {
			t.Fatal(err)
		}
		got := mustGet(t, uk, rec.Host)
		if got.Status != "landing" || got.HoldOp != "mv_acme" {
			t.Fatalf("pass %d: the uk row is %q held by %q, want landing held by mv_acme", pass, got.Status, got.HoldOp)
		}
		if got.DBPath != DefaultDBPath("/data", rec.Host) || got.Socket != DefaultSocket(rec.Host) {
			t.Fatalf("pass %d: the held row is not flat: db %q socket %q", pass, got.DBPath, got.Socket)
		}
	}
	if len(wrote.hosts) != 1 || wrote.hosts[0] != rec.Host {
		t.Fatalf("hold writes reported = %v, want the one mint", wrote.hosts)
	}

	// The ie box: the source's moving row, held by the move.
	ieRow := RouteFor(acmeRecord(sqid), "/data", `["seed"]`, false, nil)
	ieRow.Status, ieRow.HoldOp = "moving", "mv_acme"
	ie := regWith(t, ieRow)
	keep := true
	ieCfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data",
		KeepParticipant: func(host string) bool { return keep && host == rec.Host }}
	if err := Reconcile(ctx, s, ie, ieCfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, ie, rec.Host); got.Status != "moving" || got.HoldOp != "mv_acme" {
		t.Fatalf("the participant's row changed: %+v", got)
	}
	keep = false
	if err := Reconcile(ctx, s, ie, ieCfg); err != nil {
		t.Fatal(err)
	}
	if rt, err := ie.Get(rec.Host); err != nil || rt != nil {
		t.Fatalf("without the participant answer the row was not removed: %+v %v", rt, err)
	}
}

// A held mint goes through the claim hook as a ready mint does (claim Plan
// 6b): with claims on, the row is minted held under the owner the hook
// grants, flat, and a later pass over it keeps it held; a held mint the box
// refuses (no inbound, the config gate) never reaches the hook. The hold and
// the claim both have to allow the mint.
func TestReconcileHeldMintGoesThroughTheClaimHook(t *testing.T) {
	hold := func(host string) func(Record) (string, string, bool) {
		return func(r Record) (string, string, bool) {
			if r.Host == host {
				return "landing", "mv_acme", true
			}
			return "", "", false
		}
	}
	t.Run("minted held under the hook's owner", func(t *testing.T) {
		s, reg, host, cfg := instanceClaimFixture(t)
		fx := newClaimEffects(t)
		var wrote holdLog
		cfg.ClaimInstall, cfg.Hold, cfg.HoldWritten = fx.hook(t), hold(host), wrote.written
		cfg.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }
		for pass := 1; pass <= 2; pass++ {
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			got := mustGet(t, reg, host)
			if got.Status != "landing" || got.HoldOp != "mv_acme" {
				t.Fatalf("pass %d: the row is %q held by %q, want landing held by mv_acme", pass, got.Status, got.HoldOp)
			}
			if got.ClaimOwnership != fx.owner {
				t.Fatalf("pass %d: the held row's ownership is %+v, want the hook's %+v", pass, got.ClaimOwnership, fx.owner)
			}
			if got.DBPath != DefaultDBPath(cfg.DataDir, host) || got.Socket != DefaultSocket(host) {
				t.Fatalf("pass %d: the held row is not flat: db %q socket %q", pass, got.DBPath, got.Socket)
			}
		}
		fx.reached(t, 2, "the held mint, then the kept row")
		if len(wrote.hosts) != 1 || wrote.hosts[0] != host {
			t.Fatalf("hold writes reported = %v, want the one mint", wrote.hosts)
		}
	})
	t.Run("a refused held mint never reaches the hook", func(t *testing.T) {
		refuse := errors.New("no sealed config")
		cases := []struct {
			name  string
			apply func(*ReconcileConfig)
		}{
			{"inbound none", func(cfg *ReconcileConfig) { cfg.Inbound = InboundNone }},
			{"config gate", func(cfg *ReconcileConfig) {
				cfg.ConfigGate = func(context.Context, registry.Route) error { return refuse }
			}},
		}
		for _, tc := range cases {
			t.Run(tc.name, func(t *testing.T) {
				s, reg, host, cfg := instanceClaimFixture(t)
				fx := newClaimEffects(t)
				cfg.ClaimInstall, cfg.Hold = fx.hook(t), hold(host)
				tc.apply(&cfg)
				if err := Reconcile(ctx, s, reg, cfg); err != nil {
					t.Fatal(err)
				}
				fx.untouched(t, tc.name)
				if got, err := reg.Get(host); err != nil || got != nil {
					t.Fatalf("a refused held mint wrote a route: %+v %v", got, err)
				}
			})
		}
	})
}

// G10: Upsert cannot hold an existing row, so Hold on an unheld row of this
// pass's own is SetHold, and reported; a row already held by the same op is
// left as is, unreported; a row held by another op is left alone; a row this
// pass does not own is not taken into a move.
func TestReconcileHoldsAnExistingRow(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	owned := RouteFor(rec, "/data", `["seed"]`, false, nil)
	cases := []struct {
		name               string
		row                registry.Route
		wantStatus, wantOp string
		reported           bool
	}{
		{"unheld, owned", owned, "landing", "mv_acme", true},
		{"held by this op", withHold(owned, "landing", "mv_acme"), "landing", "mv_acme", false},
		{"held by another op", withHold(owned, "restoring", "rb_other"), "restoring", "rb_other", false},
		{"not this pass's row", registry.Route{Host: rec.Host, Kind: "instance", App: "other", Status: "ready", Addr: "127.0.0.1:9"}, "ready", "", false},
		// A recovery hold is written by the operation itself, which opens its
		// own row while the record still names the recovery (§7.4 reverting
		// probed, §7.8 opened): an unheld row there is one it opened.
		{"unheld, a recovery hold", owned, "ready", "", false},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			reg := regWith(t, c.row)
			var wrote holdLog
			status := "landing"
			if strings.Contains(c.name, "recovery") {
				status = "restoring"
			}
			cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
				Hold: func(r Record) (string, string, bool) { return status, r.Move.ID, r.Move.State == MoveLanding }}
			if err := Reconcile(ctx, s, reg, cfg); err != nil {
				t.Fatal(err)
			}
			got := mustGet(t, reg, rec.Host)
			if got.Status != c.wantStatus || got.HoldOp != c.wantOp {
				t.Fatalf("row is %q held by %q, want %q held by %q", got.Status, got.HoldOp, c.wantStatus, c.wantOp)
			}
			if reported := len(wrote.hosts) > 0; reported != c.reported {
				t.Fatalf("hold write reported = %v, want %v", wrote.hosts, c.reported)
			}
		})
	}
}

func withHold(rt registry.Route, status, op string) registry.Route {
	rt.Status, rt.HoldOp = status, op
	return rt
}

// A dry run reports a held mint and writes nothing.
func TestReconcileDryRunWritesNoHold(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t)
	var wrote holdLog
	cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", DryRun: true, HoldWritten: wrote.written,
		Hold: func(Record) (string, string, bool) { return "landing", "mv_acme", true }}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if rt, _ := reg.Get(rec.Host); rt != nil || len(wrote.hosts) != 0 {
		t.Fatalf("a dry run wrote: %+v %v", rt, wrote.hosts)
	}
}

// changingStore answers the record's key with first, then with every later
// read with later: the record moved on between the pass's listing and its
// hold write.
type changingStore struct {
	release.Store
	key          string
	reads        int
	first, later []byte
}

func (c *changingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == c.key {
		c.reads++
		body := c.later
		if c.reads == 1 {
			body = c.first
		}
		return body, fmt.Sprintf("etag-%d", c.reads), nil
	}
	return c.Store.Get(ctx, key)
}

// A hold write on an existing row re-reads the record first: the target
// opens its row right after its commit, and the pass's listing (read before
// the routes) can still say landing. A record that has moved on is not held
// again.
func TestReconcileHoldRereadsTheRecordBeforeHoldingARow(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	committed := rec
	m := *rec.Move
	m.State = MoveCommitted
	committed.Move = &m
	cs := &changingStore{Store: s, key: RecordKey(rec.Account, rec.App, rec.Host), first: mustJSON(t, rec), later: mustJSON(t, committed)}
	reg := regWith(t, RouteFor(rec, "/data", `["seed"]`, false, nil)) // the target's row, opened
	var wrote holdLog
	cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
		Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }}
	if err := Reconcile(ctx, cs, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if cs.reads < 2 {
		t.Fatalf("the record was read %d time(s); the hold write did not re-read it", cs.reads)
	}
	if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 0 {
		t.Fatalf("an opened row was held again from a stale listing: %+v %v", got, wrote.hosts)
	}
}

// Fix round 1 (task-6-review, Important 2): the hold write on an existing
// row is SetHoldIf, conditional on the row still being unheld in the same
// statement. Another op that takes the row after the pass's listing (here
// while the pass is paused just before its write) keeps it: the stale write
// does not land and nothing is reported.
func TestReconcileHoldWriteIsRefusedWhenAnotherOpTookTheRow(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, RouteFor(rec, "/data", `["seed"]`, false, nil)) // unheld at the listing
	var wrote holdLog
	cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
		Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }}
	paused := 0
	cfg.beforeHoldWrite = func(host string) {
		paused++
		// A rollback's hold, from a writer outside this pass's lock (root,
		// or another process): the row is another op's by the time the
		// pass writes.
		if err := reg.SetHold(host, "restoring", "rb_other"); err != nil {
			t.Error(err)
		}
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if paused != 1 {
		t.Fatalf("the pass paused %d time(s) before a hold write, want once", paused)
	}
	if got := mustGet(t, reg, rec.Host); got.Status != "restoring" || got.HoldOp != "rb_other" {
		t.Fatalf("the stale hold write landed over another op's: %+v", got)
	}
	if len(wrote.hosts) != 0 {
		t.Fatalf("a refused hold write was reported: %v", wrote.hosts)
	}
}

// Fix round 1 (task-6-review, Important 2): the hold decision, from the
// record re-read through the write, runs under the host's hold lock, the
// one the landing's open and finalisation take (WithHostHoldLock). An open
// that arrives inside that window waits for the write, then finds the hold
// under its own op and releases it; it can never run between the re-read
// and the write, so the pass cannot re-hold a row the move has opened. A
// later pass sees the committed record and leaves the opened row alone.
func TestReconcileHoldWriteIsSerialisedWithTheMoveOpening(t *testing.T) {
	s, sqid := setup(t)
	rec := movingRecord(sqid)
	if err := CreateRecord(ctx, s, rec); err != nil {
		t.Fatal(err)
	}
	reg := regWith(t, RouteFor(rec, "/data", `["seed"]`, false, nil))
	var wrote holdLog
	cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
		Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }}
	opened := make(chan struct{})
	cfg.beforeHoldWrite = func(host string) {
		// The target's open, racing the write: the commit (record CAS) and
		// the conditional release, under the same lock.
		go func() {
			defer close(opened)
			WithHostHoldLock(host, func() {
				fresh, etag, err := GetRecordETag(ctx, s, rec.Account, rec.App, rec.Host)
				if err != nil {
					t.Error(err)
					return
				}
				m := *fresh.Move
				m.State = MoveCommitted
				fresh.Move = &m
				if _, err := s.Put(ctx, RecordKey(rec.Account, rec.App, rec.Host), mustJSON(t, fresh), etag); err != nil {
					t.Error(err)
				}
				if ok, err := reg.ReleaseHold(host, rec.Move.ID); err != nil || !ok {
					t.Errorf("the open found no hold under its own op to release: %v %v", ok, err)
				}
			})
		}()
		select {
		case <-opened:
			t.Error("the open ran inside the pass's hold decision")
		case <-time.After(100 * time.Millisecond):
		}
	}
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	select {
	case <-opened:
	case <-time.After(10 * time.Second):
		t.Fatal("the open never got the lock after the pass")
	}
	if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" {
		t.Fatalf("after the open the row is %+v, want ready and unheld", got)
	}
	if len(wrote.hosts) != 1 {
		t.Fatalf("hold writes reported = %v, want the pass's one write before the open", wrote.hosts)
	}
	cfg.beforeHoldWrite = func(string) { t.Error("a later pass tried to hold the opened row") }
	if err := Reconcile(ctx, s, reg, cfg); err != nil {
		t.Fatal(err)
	}
	if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 1 {
		t.Fatalf("a later pass re-held the opened row: %+v %v", got, wrote.hosts)
	}
}

// WithHostHoldLock is exclusive per host and independent across hosts.
func TestWithHostHoldLockIsExclusivePerHost(t *testing.T) {
	entered, release := make(chan struct{}), make(chan struct{})
	go WithHostHoldLock("a.go.tito.io", func() { close(entered); <-release })
	<-entered
	WithHostHoldLock("b.go.tito.io", func() {}) // another host's lock is free
	second := make(chan struct{})
	go func() { WithHostHoldLock("a.go.tito.io", func() {}); close(second) }()
	select {
	case <-second:
		t.Fatal("a second holder entered the host's lock")
	case <-time.After(50 * time.Millisecond):
	}
	close(release)
	select {
	case <-second:
	case <-time.After(5 * time.Second):
		t.Fatal("the lock was not handed on")
	}
}

// Reconcile's own reads carry the runs to the box (run spec §5.2): every
// in-scope record with a run, including one whose row this very pass
// mints, and none out of scope.
func TestReconcileHandsRunsToRunSeen(t *testing.T) {
	s, sqid := setup(t)
	shopRecord := func(host, slug string) Record {
		rec := acmeRecord(sqid)
		rec.Host, rec.Slug, rec.App, rec.Pool = host, slug, "shop", "shared"
		return rec
	}
	withRun := shopRecord("acme.example.com", "acme")
	withRun.ActiveRun = &RunState{ID: "run_abcdefghijklmnopqrstuvwxyz", State: RunQueued, Deadline: time.Now().Add(time.Minute)}
	plain := shopRecord("plain.example.com", "plain")
	away := shopRecord("away.example.com", "away")
	away.Region = "uk"
	away.ActiveRun = withRun.ActiveRun
	for _, rec := range []Record{withRun, plain, away} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	var seen []string
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"shared"}, DataDir: "/data", RunSeen: func(rec Record) {
		seen = append(seen, rec.Host+":"+rec.ActiveRun.ID)
	}}
	if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
		t.Fatal(err)
	}
	if len(seen) != 1 || seen[0] != "acme.example.com:run_abcdefghijklmnopqrstuvwxyz" {
		t.Fatalf("RunSeen saw %q", seen)
	}
}

// strictShopRecord is a record of app "shop" under sqid, on a host under
// example.com, for the strict-listing tests below.
func strictShopRecord(sqid, slug string) Record {
	return Record{Schema: 1, Host: slug + ".shop.example.com", Slug: slug, App: "shop",
		Account: sqid, Region: "ie", Channel: "stable", CreatedAt: time.Now()}
}

func TestListAllRecordsStrictNamesUnreadableHosts(t *testing.T) {
	s, sqid := shopSetup(t)
	for _, slug := range []string{"good", "garbled", "liar"} {
		if err := CreateRecord(ctx, s, strictShopRecord(sqid, slug)); err != nil {
			t.Fatal(err)
		}
	}
	putReplace(t, s, RecordKey(sqid, "shop", "garbled.shop.example.com"), []byte("{not json"))
	// A body that decodes but names another host than its path does.
	liar := strictShopRecord(sqid, "other")
	putReplace(t, s, RecordKey(sqid, "shop", "liar.shop.example.com"), mustJSON(t, liar))

	recs, unreadable, err := ListAllRecordsStrict(ctx, s)
	if err != nil {
		t.Fatal(err)
	}
	if len(recs) != 1 || recs[0].Host != "good.shop.example.com" {
		t.Fatalf("recs %+v, want only good.shop.example.com", recs)
	}
	if len(unreadable) != 2 || !unreadable["garbled.shop.example.com"] || !unreadable["liar.shop.example.com"] {
		t.Fatalf("unreadable %v, want garbled and liar keyed by their path hosts", unreadable)
	}

	// The wrappers behave as they always did: the same records, no
	// unreadable host, no error.
	plain, err := ListAllRecords(ctx, s)
	if err != nil || len(plain) != 1 || plain[0].Host != "good.shop.example.com" {
		t.Fatalf("ListAllRecords = %+v, %v", plain, err)
	}
	tol, complete, err := ListAllRecordsTolerant(ctx, s)
	if err != nil || !complete || len(tol) != 1 || tol[0].Host != "good.shop.example.com" {
		t.Fatalf("ListAllRecordsTolerant = %+v, complete %v, %v", tol, complete, err)
	}
}

// The tolerant reader fills the same set. No wrapper exposes it, so this
// reads readAllRecords directly: a record that reads but does not decode is
// skipped, named, and does not make the set incomplete (it was read).
func TestReadAllRecordsTolerantNamesUnreadableHosts(t *testing.T) {
	s, sqid := shopSetup(t)
	for _, slug := range []string{"good", "garbled"} {
		if err := CreateRecord(ctx, s, strictShopRecord(sqid, slug)); err != nil {
			t.Fatal(err)
		}
	}
	putReplace(t, s, RecordKey(sqid, "shop", "garbled.shop.example.com"), []byte("{not json"))
	recs, complete, unreadable, err := readAllRecords(ctx, s, false)
	if err != nil || !complete || len(recs) != 1 || recs[0].Host != "good.shop.example.com" {
		t.Fatalf("recs %+v complete %v err %v", recs, complete, err)
	}
	if len(unreadable) != 1 || !unreadable["garbled.shop.example.com"] {
		t.Fatalf("unreadable %v, want garbled.shop.example.com", unreadable)
	}
}

func TestListAllRecordsStrictCleanSetHasNoUnreadableHosts(t *testing.T) {
	s, sqid := shopSetup(t)
	if err := CreateRecord(ctx, s, strictShopRecord(sqid, "one")); err != nil {
		t.Fatal(err)
	}
	recs, unreadable, err := ListAllRecordsStrict(ctx, s)
	if err != nil || len(recs) != 1 {
		t.Fatalf("recs %+v err %v", recs, err)
	}
	if unreadable == nil || len(unreadable) != 0 {
		t.Fatalf("unreadable = %#v, want an empty non-nil map", unreadable)
	}
}

// A read failure is still an error, with no records and no map: the strict
// listing adds to ListAllRecords' answer and never weakens it.
func TestListAllRecordsStrictKeepsFailingClosed(t *testing.T) {
	inner, sqid := shopSetup(t)
	if err := CreateRecord(ctx, inner, strictShopRecord(sqid, "one")); err != nil {
		t.Fatal(err)
	}
	s := &bucketShaped{Store: inner, failList: RecordsPrefix(sqid, "shop")}
	recs, unreadable, err := ListAllRecordsStrict(ctx, s)
	if err == nil || recs != nil || unreadable != nil {
		t.Fatalf("a failed app listing = %+v, %v, %v; want an error and nothing else", recs, unreadable, err)
	}
	fs := &failStore{Store: inner}
	fs.failReadsFor(RecordKey(sqid, "shop", "one.shop.example.com"))
	recs, unreadable, err = ListAllRecordsStrict(ctx, fs)
	if err == nil || recs != nil || unreadable != nil {
		t.Fatalf("a record that will not Get = %+v, %v, %v; want an error and nothing else", recs, unreadable, err)
	}
}