package console

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// mustTestSecretsKey is Task 9's (regionconfig_test.go).
var testKeyPub, testKeyFP = mustTestSecretsKey()

func bp(b bool) *bool { return &b }

type edgeOpts struct {
	hib, shared, inbound *bool // nil = the writer predates the field
	fp                   string
}

var eligibleBox = edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true), fp: testKeyFP}

// edgeJSON writes one box's edges file the way internal/host does.
func edgeJSON(box, eip, awsRegion, placement string, o edgeOpts) string {
	m := map[string]any{"v": 1, "box": box, "eip": eip, "region": awsRegion, "hosts": map[string]any{}}
	if placement != "" {
		m["placement"] = placement
	}
	if o.hib != nil {
		m["hibernation_configured"] = *o.hib
	}
	if o.shared != nil {
		m["serves_shared_pool"] = *o.shared
	}
	if o.inbound != nil {
		m["accepts_inbound"] = *o.inbound
	}
	if o.fp != "" {
		m["secrets_key_fp"] = o.fp
	}
	b, _ := json.Marshal(m)
	return string(b)
}

var (
	homeBox      = edgeJSON("ie-box", "5.6.7.8", "eu-west-1", "", eligibleBox)
	ukEligible   = edgeJSON("uk-box", "1.2.3.4", "eu-west-2", "uk", eligibleBox)
	ukNoKey      = edgeJSON("uk-box", "1.2.3.4", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)})
	ukWrongKey   = edgeJSON("uk-box", "1.2.3.4", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true), fp: "sha256:" + strings.Repeat("0", 64)})
	ukOldWriter  = edgeJSON("uk-box", "1.2.3.4", "eu-west-2", "uk", edgeOpts{})
	auPoolOnly   = edgeJSON("au-box", "9.9.9.9", "ap-southeast-2", "au", edgeOpts{hib: bp(true), shared: bp(false), inbound: bp(true), fp: testKeyFP})
	usNoInbound  = edgeJSON("us-box", "8.8.4.4", "us-east-1", "us", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(false), fp: testKeyFP})
	nzNoHosting  = edgeJSON("nz-box", "6.6.6.6", "ap-southeast-6", "nz", edgeOpts{hib: bp(false), shared: bp(true), inbound: bp(true), fp: testKeyFP})
	jpNoEIP      = edgeJSON("jp-box", "", "ap-northeast-1", "jp", eligibleBox)
	fleetBoxFile = edgeJSON("fl-box", "4.4.4.4", "eu-west-2", "fleet/acme", eligibleBox)
	// ukKeylessSibling is a second uk box reconcile would admit (shared pool,
	// inbound) that holds no key: the F2 hazard beside ukEligible.
	ukKeylessSibling = edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)})
)

// Partial reads (re-review r1): automatic non-home eligibility needs
// COMPLETE evidence, the edges snapshot Complete and every listed box's
// status read succeeded. Otherwise a failing sibling can simply drop out
// of the fold (its edges GET failed) or read as not fresh (its status GET
// failed), and the region becomes eligible through the box that was read.
// On any incomplete read no non-home region is eligible for that request
// (whole-request scope, the stricter choice); the home stays as it is.
func TestIncompleteReadsNeverMakeARegionEligible(t *testing.T) {
	ctx := context.Background()
	auBox := edgeJSON("au-box", "9.9.9.9", "ap-southeast-2", "au", eligibleBox)
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukEligible, "uk-box-2": ukKeylessSibling, "au-box": auBox} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
		setBoxSnapshot(t, s, key, time.Now())
	}
	check := func(name string, store release.Store, wantAU bool) {
		t.Helper()
		snap, err := readLiveEdgesSnapshot(ctx, store, time.Now())
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		cands := regionCandidates(snap, "ie", testKeyFP)
		uk, _ := findCandidate(cands, "uk")
		au, _ := findCandidate(cands, "au")
		home, _ := findCandidate(cands, "")
		if uk.Eligible {
			t.Errorf("%s: uk = %+v, want ineligible (a keyless sibling was listed)", name, uk)
		}
		if au.Eligible != wantAU {
			t.Errorf("%s: au eligible = %v, want %v", name, au.Eligible, wantAU)
		}
		if !home.Eligible {
			t.Errorf("%s: home = %+v, want eligible regardless", name, home)
		}
	}
	// Control: everything read. uk fails on its keyless sibling, au is clean.
	check("complete", s, true)
	// Hole 1: the keyless sibling's EDGES object cannot be read. It drops out
	// of the fold, and only the keyed box remains to speak for uk.
	check("edges GET failure", failingGet{Store: s, key: "control/edges/uk-box-2.json"}, false)
	// Hole 2: the keyless sibling's STATUS snapshot cannot be read. Unknown
	// liveness is not staleness: it must not be excluded as stale.
	check("status GET failure", failingGet{Store: s, key: "status/boxes/uk-box-2.json"}, false)
	// A corrupt status body is an incomplete read too (loadBoxes skips it).
	corrupt := release.NewDirStore(t.TempDir())
	for _, key := range []string{"ie-box", "uk-box", "uk-box-2", "au-box"} {
		body, _, err := s.Get(ctx, "control/edges/"+key+".json")
		if err != nil {
			t.Fatal(err)
		}
		if _, err := corrupt.Put(ctx, "control/edges/"+key+".json", body, ""); err != nil {
			t.Fatal(err)
		}
		setBoxSnapshot(t, corrupt, key, time.Now())
	}
	if _, err := corrupt.Put(ctx, "status/boxes/uk-box-2.json", []byte("{not json"), mustEtag(t, corrupt, "status/boxes/uk-box-2.json")); err != nil {
		t.Fatal(err)
	}
	check("corrupt status snapshot", corrupt, false)
}

// Unmatched boxes, ruling (A) (re-review r2): a FRESH status snapshot
// whose box has no edges file is incomplete evidence, whole-request, the
// same as a failed read. Reconcile does not need the edges file to admit a
// box, so a keyless live box whose edges publication failed would
// otherwise never reach the fold while its keyed sibling made the region
// eligible. Matched by the status Box.Label against the edges file's own
// box field, the pair edgesfile.EdgeIPs joins on. A STALE snapshot with no
// edges file is a retired box and is ignored.
func TestFreshStatusWithoutAnEdgesFileIsIncompleteEvidence(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukEligible} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
		setBoxSnapshot(t, s, key, time.Now())
	}
	check := func(name string, wantUK bool) {
		t.Helper()
		snap, err := readLiveEdgesSnapshot(ctx, s, time.Now())
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		cands := regionCandidates(snap, "ie", testKeyFP)
		uk, _ := findCandidate(cands, "uk")
		home, _ := findCandidate(cands, "")
		if uk.Eligible != wantUK {
			t.Errorf("%s: uk = %+v, want eligible %v", name, uk, wantUK)
		}
		if !home.Eligible {
			t.Errorf("%s: home = %+v, want eligible regardless", name, home)
		}
	}
	check("control: every fresh box has an edges file", true)
	// Box B reports status, fresh, and has published no edges file.
	setBoxSnapshot(t, s, "uk-box-2", time.Now())
	check("fresh status-only box", false)
	// The same box gone dark: a retired box, ignored.
	setBoxSnapshot(t, s, "uk-box-2", time.Now().Add(-time.Hour))
	check("stale status-only box", true)
}

// Unmatched boxes, ruling (B) REVERSED (re-review r2, production
// evidence): on the flagship every box has both objects, and a retired
// box leaves both behind, so it reads as a STALE snapshot, never an absent
// one. An edges file whose box has no status object at all is therefore
// not a retired box but unknown liveness, and it makes the evidence
// incomplete, whole-request, like (A). A stale snapshot beside the file is
// the retired box: not fresh, counts for neither side, and the region
// stays eligible through its fresh keyed sibling (the F1 case).
func TestEdgesFileWithoutAnyStatusSnapshotIsIncompleteEvidence(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukEligible} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
		setBoxSnapshot(t, s, key, time.Now())
	}
	if _, err := s.Put(ctx, "control/edges/uk-box-2.json", []byte(ukKeylessSibling), ""); err != nil {
		t.Fatal(err)
	}
	snap, err := readLiveEdgesSnapshot(ctx, s, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if snap.EvidenceComplete() {
		t.Fatalf("a keyless edges file with NO status object left the evidence complete (edges %v, boxes %v)", snap.Complete, snap.BoxesComplete)
	}
	if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); c.Eligible {
		t.Fatalf("uk = %+v, want ineligible: a box with no status object has unknown liveness", c)
	}
	if h, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), ""); !h.Eligible {
		t.Fatalf("home = %+v, want eligible regardless", h)
	}
	// The retired box: the same edges file beside a STALE snapshot. F1's
	// case, and it must keep working.
	setBoxSnapshot(t, s, "uk-box-2", time.Now().Add(-time.Hour))
	snap, err = readLiveEdgesSnapshot(ctx, s, time.Now())
	if err != nil {
		t.Fatal(err)
	}
	if !snap.EvidenceComplete() {
		t.Fatalf("a stale snapshot beside its edges file made the evidence incomplete (edges %v, boxes %v); a retired box is complete evidence", snap.Complete, snap.BoxesComplete)
	}
	if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); !c.Eligible {
		t.Fatalf("retired keyless sibling: uk = %+v, want eligible through the fresh keyed box", c)
	}
}

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

// Incomplete evidence is said once, like the other liveness diagnostics.
func TestIncompleteEvidenceIsLoggedOnce(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukEligible} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
		setBoxSnapshot(t, s, key, time.Now())
	}
	var buf bytes.Buffer
	restore := captureLogTo(&buf)
	defer restore()
	livenessSaid.Delete("incomplete")
	for i := 0; i < 2; i++ {
		snap, err := readLiveEdgesSnapshot(ctx, failingGet{Store: s, key: "status/boxes/uk-box.json"}, time.Now())
		if err != nil {
			t.Fatal(err)
		}
		if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); c.Eligible {
			t.Fatalf("round %d: uk eligible on an incomplete read", i)
		}
	}
	if n := strings.Count(buf.String(), "incomplete"); n != 1 {
		t.Fatalf("incomplete evidence logged %d times, want once: %s", n, buf.String())
	}
}

// setEdges creates OR replaces edges files, and gives each box a FRESH
// status snapshot: automatic eligibility needs one (F1), and the fixture
// boxes are all meant to be live unless a test says otherwise through
// setBoxSnapshot. putEdges is create-only (an empty ifMatch on DirStore and
// S3), so it cannot change a box's file.
func setEdges(t *testing.T, fx *consoleFixture, files map[string]string) {
	t.Helper()
	ctx := context.Background()
	for name := range files {
		setBoxSnapshot(t, fx.rel, name, time.Now())
	}
	for name, body := range files {
		key := "control/edges/" + name + ".json"
		etag := ""
		if _, cur, err := fx.rel.Get(ctx, key); err == nil {
			etag = cur
		} else if !errors.Is(err, os.ErrNotExist) {
			t.Fatal(err)
		}
		if _, err := fx.rel.Put(ctx, key, []byte(body), etag); err != nil {
			t.Fatalf("set %s: %v", key, err)
		}
	}
}

// setBoxSnapshot writes (or rewrites) a box's status/boxes/<label>.json
// generated at the given time: now for a live box, an hour ago for one that
// has stopped reporting (status.BoxStale's rule).
func setBoxSnapshot(t *testing.T, s release.Store, label string, at time.Time) {
	t.Helper()
	if err := status.WriteBox(context.Background(), s, status.Box{Label: label, GeneratedAt: at}); err != nil {
		t.Fatalf("snapshot %s: %v", label, err)
	}
}

// retireBox is what a retired box looks like in production (ruling (B)
// reversed, re-review r2): its status snapshot lingers and goes STALE, and
// here its edges file is gone too. Deleting the edges object alone would
// leave a fresh snapshot with no file, which reads as a live box that has
// not published (incomplete evidence), not as an absent region.
func retireBox(t *testing.T, fx *consoleFixture, label string) {
	t.Helper()
	if err := fx.rel.Delete(context.Background(), "control/edges/"+label+".json"); err != nil && !errors.Is(err, os.ErrNotExist) {
		t.Fatal(err)
	}
	setBoxSnapshot(t, fx.rel, label, time.Now().Add(-time.Hour))
}

type failingList struct{ release.Store }

func (failingList) List(context.Context, string) ([]string, error) {
	return nil, errors.New("bucket unreachable")
}

func captureLogTo(w io.Writer) (restore func()) {
	prev, flags := log.Writer(), log.Flags()
	log.SetOutput(w)
	return func() { log.SetOutput(prev); log.SetFlags(flags) }
}

// views is a COMPLETE snapshot of the given edges-file bodies (every file
// and every status snapshot read), every box FRESH (F1): the pure fold is
// handed what readLiveEdgesSnapshot would produce for a fleet that is all
// reporting and all readable. markStale takes one box back.
func views(t *testing.T, bodies ...string) edgesSnapshot {
	t.Helper()
	snap := edgesSnapshot{Complete: true, BoxesComplete: true}
	for _, b := range bodies {
		var f edgesFilePlacementView
		if err := json.Unmarshal([]byte(b), &f); err != nil {
			t.Fatal(err)
		}
		f.Fresh = true
		snap.Files = append(snap.Files, f)
	}
	return snap
}

// markStale is views' box that has stopped reporting: its edges file is
// still in the snapshot, its status snapshot is not fresh.
func markStale(snap edgesSnapshot, box string) edgesSnapshot {
	for i := range snap.Files {
		if snap.Files[i].Box == box {
			snap.Files[i].Fresh = false
		}
	}
	return snap
}

// F1: a retired box's edges file lingers (nothing deletes it), so the file
// alone is not liveness. A non-home box counts toward automatic eligibility
// only while its status snapshot is fresh, the rule edgesfile.EdgeIPs uses.
// The existing disappearance test deletes the object; this one leaves it.
func TestRegionCandidatesNeedAFreshBoxSnapshot(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukEligible} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
	}
	now := time.Now()
	setBoxSnapshot(t, s, "ie-box", now) // the home box reports too; see ruling (B) reversed
	uk := func(name string) regionCandidate {
		t.Helper()
		snap, err := readLiveEdgesSnapshot(ctx, s, now)
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk")
		return c
	}
	// No snapshot at all: unknown liveness (incomplete evidence, ruling (B)
	// reversed), so not eligible either way, and nothing live publishes uk.
	if c := uk("no snapshot"); c.Eligible || c.unavailableReason() != "absent" {
		t.Fatalf("no snapshot at all: uk = %+v, want absent", c)
	}
	setBoxSnapshot(t, s, "uk-box", now.Add(-time.Hour))
	if c := uk("stale"); c.Eligible || c.unavailableReason() != "absent" {
		t.Fatalf("stale snapshot, edges file still present: uk = %+v, want absent", c)
	}
	setBoxSnapshot(t, s, "uk-box", now)
	if c := uk("fresh"); !c.Eligible {
		t.Fatalf("fresh snapshot: uk = %+v, want eligible", c)
	}
	// The home candidate stays as it is: eligible throughout, exempt from
	// every eligibility fact.
	snap, _ := readLiveEdgesSnapshot(ctx, s, now)
	if h, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), ""); !h.Eligible || h.AWSRegion != "eu-west-1" {
		t.Fatalf("home = %+v", h)
	}
}

// F1 leaves the explicit door where it was: R9's keyless proof is read
// from the files, like the armed gate it sits beside, so a stale keyless
// box still proves its region keyless (C31 on an explicit --region) while
// automatic placement reads the same region as absent.
func TestStaleBoxesStillCountForTheKeylessProof(t *testing.T) {
	snap := markStale(views(t, homeBox, ukNoKey), "uk-box")
	c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk")
	if c.Eligible || c.unavailableReason() != "absent" || !c.NoKeyProven {
		t.Fatalf("stale keyless uk = %+v, want absent and proven keyless", c)
	}
	// A stale KEYED box beside a fresh keyless one: not proven, as before.
	keyless2 := edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)})
	c, _ = findCandidate(regionCandidates(markStale(views(t, homeBox, ukEligible, keyless2), "uk-box"), "ie", testKeyFP), "uk")
	if c.NoKeyProven || c.Eligible {
		t.Fatalf("stale keyed beside fresh keyless = %+v, want neither proven nor eligible", c)
	}
}

// F1: snapshots that cannot be loaded make no box fresh, and the failure
// is never said twice. The once-per-process throttle lives in edgesfile
// with the read (stage B Task 10), where TestReadLiveSnapshotSaysALoad
// FailureOnce resets it; here an earlier test may already have tripped
// it, so the count is at most one (Task 10 review M10).
func TestLiveEdgesSnapshotLoadFailureIsNotFreshAndLoggedOnce(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	if _, err := s.Put(ctx, "control/edges/uk-box.json", []byte(ukEligible), ""); err != nil {
		t.Fatal(err)
	}
	setBoxSnapshot(t, s, "uk-box", time.Now())
	var buf bytes.Buffer
	restore := captureLogTo(&buf)
	defer restore()
	for i := 0; i < 2; i++ {
		snap, err := readLiveEdgesSnapshot(ctx, failingListPrefix{Store: s, prefix: "status/boxes/"}, time.Now())
		if err != nil {
			t.Fatal(err)
		}
		if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); c.Eligible {
			t.Fatalf("round %d: snapshots unloadable, uk = %+v, want ineligible", i, c)
		}
	}
	// Count the load-failure line itself. "box snapshots" alone also
	// matches the request-level incomplete-read line, which has its own
	// once-per-process latch: the test passed only when an earlier test in
	// the same process had already tripped it.
	if n := strings.Count(buf.String(), "could not load box snapshots"); n > 1 {
		t.Fatalf("load failure logged %d times, want at most once: %s", n, buf.String())
	}
}

// failingListPrefix fails the listing of ONE prefix (the status half of the
// join) and serves every other read.
type failingListPrefix struct {
	release.Store
	prefix string
}

func (f failingListPrefix) List(ctx context.Context, prefix string) ([]string, error) {
	if prefix == f.prefix {
		return nil, errors.New("status bucket unreachable")
	}
	return f.Store.List(ctx, prefix)
}

func TestRegionCandidatesNeedPositiveEligibility(t *testing.T) {
	ca := edgeJSON("ca-box", "7.7.7.7", "ca-central-1", "ca", edgeOpts{})
	cands := regionCandidates(views(t, homeBox, ukEligible, auPoolOnly, usNoInbound, ca, nzNoHosting, jpNoEIP, fleetBoxFile), "ie", testKeyFP)
	want := map[string]string{"": "", "uk": "", "au": "not-eligible", "us": "not-eligible",
		"ca": "no-secrets-key", "nz": "not-hosting", "jp": "absent"}
	if len(cands) != len(want) {
		t.Fatalf("candidates = %+v, want exactly %v (no fleet code)", cands, want)
	}
	for code, reason := range want {
		c, ok := findCandidate(cands, code)
		if !ok {
			t.Fatalf("no candidate %q", code)
		}
		if got := c.unavailableReason(); got != reason {
			t.Errorf("%q: reason %q, want %q", code, got, reason)
		}
	}
	if h, _ := findCandidate(cands, ""); !h.Home || h.AWSRegion != "eu-west-1" || !h.Eligible {
		t.Errorf("home = %+v", h)
	}
}

// Ruling F2, refined: the "no fresh box fails" side counts only boxes
// reconcile would admit for the record, which is the same region code with
// the shared pool served and inbound accepted (reconcile.go inScope and
// the CARLOS_INBOUND=none refusal). A named-pool-only or no-inbound box can
// never be handed the record, so it neither qualifies nor disqualifies the
// region. ABSENT is not the same as false: a writer that predates the
// field is on an unknown config, and reconcile's defaults (CARLOS_POOLS
// unset serves the shared pool, CARLOS_INBOUND unset accepts) admit it, so
// absent still counts as possibly admitted, and so as failing.
func TestBoxesReconcileWouldNotAdmitNeitherQualifyNorDisqualify(t *testing.T) {
	keyless := func(shared, inbound *bool) string {
		return edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: shared, inbound: inbound})
	}
	harmless := map[string]string{
		"keyless named-pool-only box": keyless(bp(false), bp(true)),
		"keyless no-inbound box":      keyless(bp(true), bp(false)),
		"keyless neither":             keyless(bp(false), bp(false)),
	}
	for name, body := range harmless {
		cands := regionCandidates(views(t, homeBox, ukEligible, body), "ie", testKeyFP)
		if c, _ := findCandidate(cands, "uk"); !c.Eligible {
			t.Errorf("%s beside a keyed eligible box: uk = %+v, want eligible (reconcile would never admit it)", name, c)
		}
		if n := eligibleCount(cands); n != 2 {
			t.Errorf("%s: %d eligible candidates, want the home and uk", name, n)
		}
	}
	// Possibly admitted, so failing: a keyless box that serves the shared
	// pool and takes inbound (the flipped mixed-box case), and one that
	// is silent on either field.
	hazard := map[string]string{
		"keyless shared-pool inbound box": keyless(bp(true), bp(true)),
		"keyless, shared pool absent":     keyless(nil, bp(true)),
		"keyless, inbound absent":         keyless(bp(true), nil),
	}
	for name, body := range hazard {
		if c, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible, body), "ie", testKeyFP), "uk"); c.Eligible {
			t.Errorf("%s beside a keyed eligible box: uk = %+v, want ineligible", name, c)
		}
	}
	// On its own, a box reconcile would not admit still makes no region
	// eligible: nothing qualifies it. Keyed, so the reason is the pool.
	poolOnlyKeyed := edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(false), inbound: bp(true), fp: testKeyFP})
	if c, _ := findCandidate(regionCandidates(views(t, homeBox, poolOnlyKeyed), "ie", testKeyFP), "uk"); c.Eligible || c.unavailableReason() != "not-eligible" {
		t.Errorf("pool-only box alone: uk = %+v, want not-eligible", c)
	}
}

// Task 10 (final review, MUST FIX): each capability field ABSENT on its
// own, and each explicitly FALSE on its own, with everything else eligible
// and the key present. The older fixtures that left fields absent also
// lacked the key, so a permissive nil (absent read as yes) would have been
// hidden behind the key check. Unknown is not yes (spec §6.1).
func TestEachCapabilityFieldAloneBlocksEligibility(t *testing.T) {
	cases := []struct {
		name   string
		o      edgeOpts
		reason string
	}{
		{"hibernation absent", edgeOpts{shared: bp(true), inbound: bp(true), fp: testKeyFP}, "not-eligible"},
		{"hibernation false", edgeOpts{hib: bp(false), shared: bp(true), inbound: bp(true), fp: testKeyFP}, "not-hosting"},
		{"shared pool absent", edgeOpts{hib: bp(true), inbound: bp(true), fp: testKeyFP}, "not-eligible"},
		{"shared pool false", edgeOpts{hib: bp(true), shared: bp(false), inbound: bp(true), fp: testKeyFP}, "not-eligible"},
		{"inbound absent", edgeOpts{hib: bp(true), shared: bp(true), fp: testKeyFP}, "not-eligible"},
		{"inbound false", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(false), fp: testKeyFP}, "not-eligible"},
	}
	for _, c := range cases {
		box := edgeJSON("uk-box", "1.2.3.4", "eu-west-2", "uk", c.o)
		cands := regionCandidates(views(t, homeBox, box), "ie", testKeyFP)
		uk, ok := findCandidate(cands, "uk")
		if !ok {
			t.Fatalf("%s: no uk candidate", c.name)
		}
		if uk.Eligible || uk.unavailableReason() != c.reason {
			t.Errorf("%s: uk = %+v (reason %q), want ineligible with reason %q", c.name, uk, uk.unavailableReason(), c.reason)
		}
		if n := eligibleCount(cands); n != 1 {
			t.Errorf("%s: %d eligible candidates, want the home alone", c.name, n)
		}
	}
	// The control: the same box with every field present and true is
	// eligible, so the cases above fail on the one field they vary.
	if uk, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible), "ie", testKeyFP), "uk"); !uk.Eligible {
		t.Fatalf("control: uk = %+v, want eligible", uk)
	}
}

// R9: a box must hold the deployment's secrets key to receive placed
// instances, and only a CURRENT writer can prove it does not.
func TestRegionCandidatesSecretsKey(t *testing.T) {
	cases := []struct {
		name        string
		body        string
		expected    string
		reason      string
		noKeyProven bool
	}{
		{"matching key", ukEligible, testKeyFP, "", false},
		{"no key, current writer", ukNoKey, testKeyFP, "no-secrets-key", true},
		{"wrong key, current writer", ukWrongKey, testKeyFP, "no-secrets-key", true},
		{"old writer: unknown", ukOldWriter, testKeyFP, "no-secrets-key", false},
		{"console cannot judge", ukEligible, "", "no-secrets-key", false},
	}
	for _, c := range cases {
		cand, _ := findCandidate(regionCandidates(views(t, homeBox, c.body), "ie", c.expected), "uk")
		if got := cand.unavailableReason(); got != c.reason || cand.NoKeyProven != c.noKeyProven {
			t.Errorf("%s: reason %q proven %v, want %q %v", c.name, got, cand.NoKeyProven, c.reason, c.noKeyProven)
		}
	}
	// One box with the key and one without: the region is not PROVEN
	// keyless (the explicit door may still use it), but it is NOT eligible
	// for automatic placement (final review F2). A record names a region,
	// not a box, and reconcile admits every box in the region that serves
	// the shared pool and takes inbound: the keyless box can win the lease
	// and wake the instance with no config, secrets or store binding.
	ukKeyless2 := edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)})
	cand, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible, ukKeyless2), "ie", testKeyFP), "uk")
	if cand.Eligible || cand.NoKeyProven || cand.unavailableReason() != "not-eligible" {
		t.Errorf("mixed region = %+v, want ineligible (not-eligible) and not proven keyless", cand)
	}
}

// F2: a region is automatically eligible only when at least one fresh box
// is fully eligible AND no fresh box publishing the code fails the
// predicate. One box reconcile would admit that cannot serve the record
// takes the region out; stale boxes count for neither side.
func TestEligibilityNeedsEveryFreshBoxToQualify(t *testing.T) {
	sibling := func(o edgeOpts) string { return edgeJSON("uk-box-2", "1.2.3.5", "eu-west-2", "uk", o) }
	keyed := func(hib, shared, inbound *bool) edgeOpts {
		return edgeOpts{hib: hib, shared: shared, inbound: inbound, fp: testKeyFP}
	}
	bad := map[string]string{
		"keyless":                 sibling(edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)}),
		"wrong key":               sibling(edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true), fp: "sha256:" + strings.Repeat("0", 64)}),
		"hibernation false":       sibling(keyed(bp(false), bp(true), bp(true))),
		"hibernation absent":      sibling(keyed(nil, bp(true), bp(true))),
		"shared pool absent":      sibling(keyed(bp(true), nil, bp(true))),
		"inbound absent":          sibling(keyed(bp(true), bp(true), nil)),
		"old writer, every field": sibling(edgeOpts{}),
		"no dial address":         edgeJSON("uk-box-2", "", "eu-west-2", "uk", eligibleBox),
	}
	// A sibling that POSITIVELY publishes serves_shared_pool false or
	// accepts_inbound false is one reconcile would never hand the record:
	// see TestBoxesReconcileWouldNotAdmitNeitherQualifyNorDisqualify.
	for name, body := range bad {
		cands := regionCandidates(views(t, homeBox, ukEligible, body), "ie", testKeyFP)
		c, _ := findCandidate(cands, "uk")
		if c.Eligible || c.unavailableReason() == "" {
			t.Errorf("%s sibling: uk = %+v, want ineligible", name, c)
		}
		if n := eligibleCount(cands); n != 1 {
			t.Errorf("%s sibling: %d eligible candidates, want the home alone", name, n)
		}
		// The diagnostics still describe the boxes that ARE there: the
		// eligible box is reachable, hosting and keyed, so the C16 word is
		// not-eligible, never absent.
		if c.unavailableReason() == "absent" {
			t.Errorf("%s sibling: uk read as absent with a reachable box present", name)
		}
	}
	// Two fully eligible boxes: eligible.
	if c, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible, sibling(eligibleBox)), "ie", testKeyFP), "uk"); !c.Eligible {
		t.Fatalf("two eligible boxes: uk = %+v, want eligible", c)
	}
	// A STALE failing sibling counts for neither side: the live box alone
	// decides, and the region stays eligible.
	if c, _ := findCandidate(regionCandidates(markStale(views(t, homeBox, ukEligible, bad["keyless"]), "uk-box-2"), "ie", testKeyFP), "uk"); !c.Eligible {
		t.Fatalf("stale keyless sibling: uk = %+v, want eligible", c)
	}
	// A stale ELIGIBLE box beside a fresh failing one lends nothing.
	if c, _ := findCandidate(regionCandidates(markStale(views(t, homeBox, ukEligible, bad["keyless"]), "uk-box"), "ie", testKeyFP), "uk"); c.Eligible || c.unavailableReason() != "no-secrets-key" {
		t.Fatalf("stale eligible, fresh keyless: uk = %+v, want no-secrets-key", c)
	}
}

// codex plan-R2 1: a negative proof needs EVERY box. With one file
// unreadable or malformed, a readable keyless box proves nothing.
func TestNoKeyProvenNeedsACompleteRead(t *testing.T) {
	incomplete := views(t, homeBox, ukNoKey)
	incomplete.Complete = false
	if c, _ := findCandidate(regionCandidates(incomplete, "ie", testKeyFP), "uk"); c.NoKeyProven {
		t.Fatal("an incomplete read proved a region keyless")
	}
	ctx := context.Background()
	for name, extra := range map[string]string{"malformed": "{not json", "unversioned": `{"box":"uk-box-2"}`} {
		s := release.NewDirStore(t.TempDir())
		for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukNoKey, "uk-box-2": extra} {
			if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
				t.Fatal(err)
			}
		}
		snap, err := readEdgesSnapshot(ctx, s)
		if err != nil || snap.Complete {
			t.Fatalf("%s: snapshot complete=%v err=%v, want incomplete", name, snap.Complete, err)
		}
		if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); c.NoKeyProven {
			t.Fatalf("%s: proved keyless with a box unassessed", name)
		}
	}
	s := release.NewDirStore(t.TempDir())
	for key, body := range map[string]string{"ie-box": homeBox, "uk-box": ukNoKey, "uk-box-2": ukEligible} {
		if _, err := s.Put(ctx, "control/edges/"+key+".json", []byte(body), ""); err != nil {
			t.Fatal(err)
		}
	}
	snap, err := readEdgesSnapshot(ctx, failingGet{Store: s, key: "control/edges/uk-box-2.json"})
	if err != nil || snap.Complete {
		t.Fatalf("failed GET: complete=%v err=%v, want incomplete", snap.Complete, err)
	}
	if c, _ := findCandidate(regionCandidates(snap, "ie", testKeyFP), "uk"); c.NoKeyProven {
		t.Fatal("a failed GET let the readable keyless box prove the region keyless")
	}
}

type failingGet struct {
	release.Store
	key string
}

func (f failingGet) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == f.key {
		return nil, "", errors.New("read timeout")
	}
	return f.Store.Get(ctx, key)
}

// codex plan-R2 2: a candidate's coordinates come from its fully eligible
// boxes only; ties among them go to the lexically smallest AWS region. The
// mixed region is no longer eligible (F2), but it is still NAMED (placeOf,
// C16's refusal) from the box that qualifies, not from the ones that do not.
func TestCandidateLocationComesFromEligibleBoxes(t *testing.T) {
	syd1 := edgeJSON("uk-syd-1", "2.2.2.1", "ap-southeast-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)})
	syd2 := edgeJSON("uk-syd-2", "2.2.2.2", "ap-southeast-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(false), fp: testKeyFP})
	c, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible, syd1, syd2), "ie", testKeyFP), "uk")
	if c.Eligible || c.AWSRegion != "eu-west-2" {
		t.Fatalf("mixed eligibility = %+v, want ineligible (F2) and located at the qualifying box's eu-west-2", c)
	}
	dub := edgeJSON("uk-dub", "2.2.2.3", "eu-west-1", "uk", eligibleBox)
	if c, _ := findCandidate(regionCandidates(views(t, homeBox, ukEligible, dub), "ie", testKeyFP), "uk"); c.AWSRegion != "eu-west-1" {
		t.Fatalf("eligible tie = %q, want the lexically smallest eu-west-1", c.AWSRegion)
	}
}

func TestRegionCandidatesFoldTheHomesName(t *testing.T) {
	coded := edgeJSON("ie2", "5.6.7.9", "eu-west-1", "ie", eligibleBox)
	cands := regionCandidates(views(t, homeBox, coded), "ie", testKeyFP)
	if len(cands) != 1 || !cands[0].Home {
		t.Fatalf("candidates = %+v, want the home alone", cands)
	}
}

// spec §6.3, §6.5: a candidate with no point is logged once per code, and
// the line carries no caller coordinates (there are none to carry).
func TestRegionCandidatesLogAMissingPointOnce(t *testing.T) {
	var buf bytes.Buffer
	restore := captureLogTo(&buf)
	defer restore()
	noPointSaid.Delete("odd") // the throttle is process-wide; -count=N reruns this test
	odd := edgeJSON("odd-box", "3.3.3.3", "xx-nowhere-1", "odd", eligibleBox)
	regionCandidates(views(t, homeBox, odd), "ie", testKeyFP)
	regionCandidates(views(t, homeBox, odd), "ie", testKeyFP)
	if n := strings.Count(buf.String(), `region "odd"`); n != 1 {
		t.Fatalf("missing-point line logged %d times, want once: %s", n, buf.String())
	}
}

func TestReadEdgesSnapshotReportsListFailure(t *testing.T) {
	if _, err := readEdgesSnapshot(context.Background(), failingList{}); err == nil {
		t.Fatal("a failed listing must be an error, not an empty answer (codex R2-2)")
	}
	snap, err := readEdgesSnapshot(context.Background(), release.NewDirStore(t.TempDir()))
	if err != nil || len(snap.Files) != 0 || !snap.Complete {
		t.Fatalf("an empty prefix = %+v %v, want complete and empty", snap, err)
	}
}

func TestAppPolicyStates(t *testing.T) {
	fx, _ := instanceFixture(t, "code-policy")
	acct := &Account{ID: OpsAccountID, Sqid: "bab"}
	ctx := context.Background()
	if p := fx.srv.appPolicy(ctx, acct, "titogo"); p.State != policyAbsent || !p.allows("uk") {
		t.Fatalf("no manifest = %+v", p)
	}
	m := titoManifest()
	m.Pool, m.Regions = "", []string{"uk"}
	enableInstances(t, fx, "titogo", m)
	if p := fx.srv.appPolicy(ctx, acct, "titogo"); p.State != policyKnown || p.allows("") || !p.allows("uk") || p.pooled() {
		t.Fatalf("known = %+v", p)
	}
	m.Pool = "tito"
	enableInstances(t, fx, "titogo", m)
	if !fx.srv.appPolicy(ctx, acct, "titogo").pooled() {
		t.Fatal("a pooled manifest must read as pooled")
	}
	if err := fx.st.SetAppPlacement(OpsAccountID, "titogo", "fleet/acme"); err != nil {
		t.Fatal(err)
	}
	if p := fx.srv.appPolicy(ctx, acct, "titogo"); p.Placed != "fleet/acme" {
		t.Fatalf("placed = %q", p.Placed)
	}
	_, etag, err := fx.rel.Get(ctx, provision.ManifestKey("bab", "titogo"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(ctx, provision.ManifestKey("bab", "titogo"), []byte("{not json"), etag); err != nil {
		t.Fatal(err)
	}
	if p := fx.srv.appPolicy(ctx, acct, "titogo"); p.State != policyUnreadable || p.allows("") {
		t.Fatalf("unreadable = %+v, want unreadable and nothing allowed", p)
	}
	fx.srv.cfg.Releases = nil
	if p := fx.srv.appPolicy(ctx, acct, "titogo"); p.State != policyUnreadable || p.allows("uk") {
		t.Fatalf("no bucket = %+v, want unreadable and nothing allowed", p)
	}
}

// splitCapability pairs two boxes in one region that between them publish
// every eligibility fact while neither publishes them all (codex
// plan-R3 4). Task 11 reuses it for the guess.
var splitCapability = map[string][2]string{
	"keyless ordinary box beside a keyed named-pool-only box": {
		edgeJSON("uk-a", "1.2.3.4", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)}),
		edgeJSON("uk-b", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(false), inbound: bp(true), fp: testKeyFP}),
	},
	"keyed no-inbound box beside a keyless inbound box": {
		edgeJSON("uk-a", "1.2.3.4", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(false), fp: testKeyFP}),
		edgeJSON("uk-b", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)}),
	},
	"keyed non-hosting box beside a keyless hosting box": {
		edgeJSON("uk-a", "1.2.3.4", "eu-west-2", "uk", edgeOpts{hib: bp(false), shared: bp(true), inbound: bp(true), fp: testKeyFP}),
		edgeJSON("uk-b", "1.2.3.5", "eu-west-2", "uk", edgeOpts{hib: bp(true), shared: bp(true), inbound: bp(true)}),
	},
}

// Eligibility is ONE box's facts, never the region's union of them.
func TestEligibilityNeverCombinesBoxes(t *testing.T) {
	for name, boxes := range splitCapability {
		cands := regionCandidates(views(t, homeBox, boxes[0], boxes[1]), "ie", testKeyFP)
		if c, _ := findCandidate(cands, "uk"); c.Eligible || c.unavailableReason() == "" {
			t.Errorf("%s: uk = %+v, want ineligible", name, c)
		}
		if n := eligibleCount(cands); n != 1 {
			t.Errorf("%s: %d eligible candidates, want the home alone", name, n)
		}
	}
}

// candsOrHome answers from this console's home and secrets key, and falls
// back to the home alone when the listing fails or there is no bucket.
func TestCandsOrHomeUsesTheServerAndFallsBackToTheHome(t *testing.T) {
	fx := newConsoleFixture(t)
	ctx := context.Background()
	fx.srv.cfg.HomeRegion, fx.srv.cfg.SecretsPub = "ie", testKeyPub
	setEdges(t, fx, map[string]string{"ie-box": homeBox, "uk-box": ukEligible, "ie2": edgeJSON("ie2", "5.6.7.9", "eu-west-1", "ie", eligibleBox)})
	cands := fx.srv.candsOrHome(ctx)
	if len(cands) != 2 || eligibleCount(cands) != 2 {
		t.Fatalf("candidates = %+v, want the home (ie folded in) and uk, both eligible", cands)
	}
	fx.srv.cfg.SecretsPub = ""
	if c, _ := findCandidate(fx.srv.candsOrHome(ctx), "uk"); c.Eligible || c.unavailableReason() != "no-secrets-key" {
		t.Fatalf("no console key: uk = %+v, want no-secrets-key", c)
	}
	for name, store := range map[string]release.Store{"failed listing": failingList{Store: fx.rel}, "no bucket": nil} {
		fx.srv.cfg.Releases = store
		if cands := fx.srv.candsOrHome(ctx); len(cands) != 1 || !cands[0].Home || !cands[0].Eligible {
			t.Errorf("%s: candidates = %+v, want the home alone", name, cands)
		}
	}
}

// O6, O7 and O13 to O15 are this file's diagnostics, taken verbatim from
// the approved copy record and compared as quoted literals.
func TestRegionCandidatesCopyMatchesTheRecord(t *testing.T) {
	raw, err := os.ReadFile("../../docs/superpowers/plans/2026-10-02-nearest-region-copy.json")
	if err != nil {
		t.Fatal(err)
	}
	var rec struct {
		Entries []struct {
			ID, File, Text string
		} `json:"entries"`
	}
	if err := json.Unmarshal(raw, &rec); err != nil {
		t.Fatal(err)
	}
	src, err := os.ReadFile("regioncandidates.go")
	if err != nil {
		t.Fatal(err)
	}
	var ids []string
	for _, e := range rec.Entries {
		if e.File != "internal/console/regioncandidates.go" {
			continue
		}
		ids = append(ids, e.ID)
		if !strings.Contains(string(src), strconv.Quote(e.Text)) {
			t.Errorf("%s: regioncandidates.go has no literal %s", e.ID, strconv.Quote(e.Text))
		}
	}
	if got := strings.Join(ids, ","); got != "O6,O7,O13,O14,O15" {
		t.Errorf("the record maps %s to regioncandidates.go, want O6,O7,O13,O14,O15", got)
	}
}

func TestEdgesViewParsesBoxEvidenceAndToleratesItsAbsence(t *testing.T) {
	ctx, s := context.Background(), release.NewDirStore(t.TempDir())
	with := `{"v":1,"box":"i-0abc","eip":"1.2.3.4","region":"eu-west-2","placement":"uk","hosts":{},"box_id":"i-0abc","capabilities":["region-ha-1"],"draining":true,"headroom":{"mem_free_mib":2210,"disk_free_mib":31200,"awake":14}}`
	without := `{"v":1,"box":"old-box","eip":"5.6.7.8","region":"eu-west-1","hosts":{}}`
	for k, b := range map[string]string{"i-0abc": with, "old-box": without} {
		if _, err := s.Put(ctx, "control/edges/"+k+".json", []byte(b), ""); err != nil {
			t.Fatal(err)
		}
	}
	snap, err := readEdgesSnapshot(ctx, s)
	if err != nil || !snap.Complete {
		t.Fatalf("snap complete=%v err=%v", snap.Complete, err)
	}
	by := map[string]edgesFilePlacementView{}
	for _, f := range snap.Files {
		by[f.Box] = f
	}
	n := by["i-0abc"]
	if n.BoxID != "i-0abc" || len(n.Capabilities) != 1 || !n.Draining || n.Headroom == nil || n.Headroom.Awake != 14 {
		t.Fatalf("new box parsed as %+v", n)
	}
	o := by["old-box"]
	if o.BoxID != "" || o.Capabilities != nil || o.Draining || o.Headroom != nil {
		t.Fatalf("old box parsed as %+v", o)
	}
}