// Task 10 of the customer-fleets plan: the two-box fleet acceptance test —
// an edge and TWO fake fleet boxes dialing REAL reverse channels
// ("pi/box-a", "pi/box-b", authenticated out of the registry), driven
// through the whole failure sequence the design spec promises:
//
//	(a) a bound route rides its bound box's channel;
//	(b) the bound box goes dark -> the survivor serves (rule 2), wakes,
//	    wins the lapsed lease, publishes a takeover binding, and the
//	    platform's ReconcileFleetBindings restamps Route.FleetBox to it;
//	(c) the dead box comes back, LOSES the reclaim (lease.HeldError) and
//	    must not steal routing back;
//	(d) rotate/detach: rotation refuses NEW dials with the old token
//	    within one fingerprint poll but does NOT evict the live session;
//	    detach closes the channel outright;
//	(e) both boxes dark -> 503 + Retry-After: 2.
//
// It is `package edge_test`, not `package edge`, and that is forced, not
// stylistic: this test has to drive the REAL binding writer and reconciler
// (internal/host), and internal/host imports internal/edge — the cycle the
// tunnelclient package's own doc comment calls out. An external test package
// may import a package that depends on the package under test, so the file
// stays where the plan puts it (internal/edge/) and pays for it only in a
// few small helpers duplicated from tunnel_e2e_test.go (startFleetBox,
// waitChannel) that cannot be shared across the package boundary.
//
// Hermetic and fast (~10s): loopback httptest edge, real h2 reverse channels
// via the production tunnelclient.Loop, a DirStore for the platform's
// deployment bucket, and an in-test model of the per-host S3 lease. Only the
// LEASE is modeled — see fakeFleetLease — because internal/lease speaks to S3
// and its short-TTL seam (acquireTTL) is package-private; the live-S3
// counterpart is TestDarkHostCannotReclaimAfterFailover in
// internal/lease/lease_test.go. Everything else here is the real thing.
package edge_test

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/host"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/tunnelclient"
)

const (
	e2eFleet   = "pi"
	e2eAppHost = "shop.example"

	// e2eDialBound is how long a fake box gets to complete its first dial (or
	// a re-dial after a reap) — the same promise tunnel_e2e_test.go's
	// reconnect bound makes, loose enough that a loaded CI box cannot flake
	// it.
	e2eDialBound = 5 * time.Second
	// e2eConvergeBound is the registry->edge convergence promise: the edge
	// polls registry.Fingerprint every 2s (Edge.Watch) and reloads routes AND
	// fleet channel auth on the same tick, so any registry write — a rotated
	// token, a detached box, a restamped FleetBox — is live within one poll.
	// 5s is that 2s plus room for a slow reload; a regression to "only on
	// restart" (or a fingerprint that misses fleet_boxes.rev) fails here.
	e2eConvergeBound = 5 * time.Second
	// e2eLeaseTTL models internal/lease's LeaseTTL, compressed from 15
	// minutes to 2 seconds so a whole go-dark -> lapse -> takeover sequence
	// fits in a test. The renewer below refreshes at a quarter of it, so a
	// box that is ALIVE holds its lease indefinitely and a box that has died
	// (its renewer stopped) lapses within one TTL — which is exactly the
	// timing property the failover depends on.
	e2eLeaseTTL = 2 * time.Second
)

// --- the modeled half: the per-host single-writer lease -----------------------

// fakeFleetLease models internal/lease's per-host S3 lease: at most one box
// holds a given host at a time, a live hold refuses everyone else with the
// SAME typed *lease.HeldError the real Acquire surfaces (the error the
// activator actually reacts to), and a hold lapses TTL after its last
// renewal. It is a model rather than the real thing because internal/lease
// talks to S3 and its short-TTL seam is package-private; what this test needs
// from it is the ARBITRATION — who may publish a takeover binding — not the
// S3 conditional-write mechanics, which lease_test.go already pins against a
// live bucket.
type fakeFleetLease struct {
	ttl  time.Duration
	mu   sync.Mutex
	held map[string]fakeLeaseHold
}

type fakeLeaseHold struct {
	box     string
	expires time.Time
}

func newFakeFleetLease(ttl time.Duration) *fakeFleetLease {
	return &fakeFleetLease{ttl: ttl, held: make(map[string]fakeLeaseHold)}
}

// acquire claims host for box, or fails with a *lease.HeldError (wrapped, as
// lease.acquireTTL wraps it) if a DIFFERENT box's hold is still live. A box
// re-acquiring its own live hold succeeds, mirroring the real leaser.
func (l *fakeFleetLease) acquire(host, box string) error {
	l.mu.Lock()
	defer l.mu.Unlock()
	if cur, ok := l.held[host]; ok && cur.box != box && time.Now().Before(cur.expires) {
		return fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: cur.box, ExpiresAt: cur.expires})
	}
	l.held[host] = fakeLeaseHold{box: box, expires: time.Now().Add(l.ttl)}
	return nil
}

// startRenewer is internal/lease's startRenewer: a live holder keeps pushing
// its expiry out until it stops. Stopping it is how this test says "that box
// just died" — no renewals, so the hold lapses one TTL later and a survivor
// can take over, which is the design spec's failure sequence exactly.
func (l *fakeFleetLease) startRenewer(host, box string) func() {
	stop := make(chan struct{})
	done := make(chan struct{})
	go func() {
		defer close(done)
		t := time.NewTicker(l.ttl / 4)
		defer t.Stop()
		for {
			select {
			case <-stop:
				return
			case <-t.C:
				l.mu.Lock()
				if cur, ok := l.held[host]; ok && cur.box == box {
					l.held[host] = fakeLeaseHold{box: box, expires: time.Now().Add(l.ttl)}
				}
				l.mu.Unlock()
			}
		}
	}()
	var once sync.Once
	return func() {
		once.Do(func() { close(stop) })
		<-done
	}
}

// --- helpers ------------------------------------------------------------------

func e2eHashToken(s string) string {
	sum := sha256.Sum256([]byte(s))
	return hex.EncodeToString(sum[:])
}

// startFleetBox runs the PRODUCTION host-side dial loop (the same one
// `carlos agent` runs on a Pi) against edgeURL under a fleet channel name,
// and returns a stop func that cancels it and waits — "this box's agent
// died". Mirrors tunnel_e2e_test.go's startFakeRemoteHost, which lives in
// `package edge` and so cannot be reached from here.
func startFleetBox(t *testing.T, edgeURL, name, token string, handler http.Handler) func() {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		tunnelclient.Loop(ctx, tunnelclient.Config{
			URL: edgeURL, Name: name, Token: token, Handler: handler,
			Logf: func(format string, args ...any) { t.Logf("fake box "+name+": "+format, args...) },
		})
	}()
	var once sync.Once
	return func() {
		once.Do(func() {
			cancel()
			wg.Wait() // no t.Logf after the test returns
		})
	}
}

// namedFleetHandler answers every request with the box's own label, so a
// response body names the box that actually served it — the only way to tell
// which channel a request rode.
func namedFleetHandler(name string) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		io.WriteString(w, name)
	})
}

func waitChannel(t *testing.T, hub *edge.TunnelHub, name, want string, within time.Duration) {
	t.Helper()
	deadline := time.Now().Add(within)
	for time.Now().Before(deadline) {
		if hub.State(name) == want {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("channel %s state = %q, want %q within %s", name, hub.State(name), want, within)
}

// e2eGet is one client request for the fleet-placed hostname.
func e2eGet(t *testing.T, edgeURL, host string) (int, string) {
	t.Helper()
	resp, err := e2eDo(edgeURL, host, 10*time.Second)
	if err != nil {
		t.Fatalf("GET %s (host %s): %v", edgeURL, host, err)
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatalf("read body: %v", err)
	}
	return resp.StatusCode, string(body)
}

func e2eDo(edgeURL, host string, timeout time.Duration) (*http.Response, error) {
	req, err := http.NewRequest("GET", edgeURL+"/", nil)
	if err != nil {
		return nil, err
	}
	req.Host = host
	c := &http.Client{Timeout: timeout}
	return c.Do(req)
}

// waitBody polls until the fleet host answers 200 with want, which is how the
// edge's convergence on a registry change (a restamped FleetBox, applied by
// the 2s Watch poll rebuilding the route's proxy) is observed from outside.
func waitBody(t *testing.T, edgeURL, host, want string, within time.Duration) {
	t.Helper()
	deadline := time.Now().Add(within)
	var lastCode int
	var lastBody string
	for time.Now().Before(deadline) {
		lastCode, lastBody = e2eGet(t, edgeURL, host)
		if lastCode == http.StatusOK && lastBody == want {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatalf("host %s answered %d %q, want 200 %q within %s", host, lastCode, lastBody, want, within)
}

// registerProbe asks the hub's /register endpoint the auth question ALONE: it
// presents a channel name and bearer token but deliberately omits the
// Connection/Upgrade headers, so an ACCEPTED credential answers 400 (the
// wire-contract gate, which sits after auth on purpose) and a REJECTED one
// answers 401. That distinction is the whole point — it reads the hub's live
// credential table without hijacking anything, so probing rotation cannot
// disturb the very session whose survival this test also asserts. A probe
// that completed a real handshake would replace the live channel and prove
// the opposite of what it measured.
func registerProbe(t *testing.T, edgeURL, hubHost, name, token string) int {
	t.Helper()
	req, err := http.NewRequest("GET", edgeURL+"/register", nil)
	if err != nil {
		t.Fatalf("build probe: %v", err)
	}
	req.Host = hubHost
	req.Header.Set("Carlos-Tunnel-Name", name)
	req.Header.Set("Authorization", "Bearer "+token)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("probe %s: %v", name, err)
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, resp.Body)
	return resp.StatusCode
}

func waitProbe(t *testing.T, edgeURL, hubHost, name, token string, want int, within time.Duration) time.Duration {
	t.Helper()
	start := time.Now()
	deadline := start.Add(within)
	var last int
	for time.Now().Before(deadline) {
		last = registerProbe(t, edgeURL, hubHost, name, token)
		if last == want {
			return time.Since(start)
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatalf("register probe for %s answered %d, want %d within %s", name, last, want, within)
	return 0
}

// --- the test -----------------------------------------------------------------

// TestTwoBoxFleetEndToEnd is the plan's Task 10 acceptance evidence. The legs
// run in sequence against ONE world (each depends on the state the previous
// one left), so it is a single function rather than subtests: a failure means
// the fleet stopped behaving at that step, and nothing after it is meaningful.
//
// Box naming is load-bearing. The fallback rule (rule 2) picks
// ConnectedOfFleet's SORTED FIRST entry, so "box-a" is what a
// fallback-only implementation would always answer:
//
//   - Leg (a) binds the route to box-b (sorts LAST) while both boxes are
//     connected — only rule 1 can answer "box-b" there.
//   - Leg (c) then re-connects box-b while the route is bound to box-a: if
//     leg (b)'s restamp had never happened (FleetBox still box-b), or if the
//     returning box had stolen the binding back, rule 1 would answer "box-b".
//     Insisting on "box-a" with BOTH boxes connected is what makes the
//     restamp and the lost reclaim visible rather than assumed.
func TestTwoBoxFleetEndToEnd(t *testing.T) {
	ctx := context.Background()
	const tokenA, tokenB = "tok-a", "tok-b"

	// --- the platform box: registry, edge, hub, 2s watcher --------------------
	// Deliberately NOT closed at cleanup: Edge.Watch may still be mid-poll when
	// the test returns (it exits within one 2s tick), and closing the registry
	// under it would race a live reload for no benefit — the temp dir goes away
	// either way.
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	if err := reg.UpsertFleet(registry.Fleet{Name: e2eFleet, Account: "amadan", Bucket: "pi-data"}); err != nil {
		t.Fatalf("UpsertFleet: %v", err)
	}
	for label, tok := range map[string]string{"box-a": tokenA, "box-b": tokenB} {
		if err := reg.AddFleetBox(registry.FleetBox{
			Fleet: e2eFleet, Label: label, TokenHash: e2eHashToken(tok), Status: "active",
		}); err != nil {
			t.Fatalf("AddFleetBox(%s): %v", label, err)
		}
	}
	// The app: placed on the fleet, bound to box-b (see the naming note above).
	if err := reg.Upsert(registry.Route{
		Host: e2eAppHost, Kind: "instance", Status: "ready",
		Region: registry.PlacementTarget(e2eFleet), FleetBox: "box-b",
	}); err != nil {
		t.Fatalf("upsert route: %v", err)
	}

	e, err := edge.New(reg, nil)
	if err != nil {
		t.Fatalf("edge.New: %v", err)
	}
	srv := httptest.NewServer(e)
	t.Cleanup(srv.Close)
	u, err := url.Parse(srv.URL)
	if err != nil {
		t.Fatal(err)
	}
	// One listener carries both app traffic and channel registration; the hub
	// answers on the listener's own hostname (production: CARLOS_TUNNEL_HOST).
	hub := edge.NewTunnelHub(u.Hostname(), nil)
	e.EnableTunnels(hub)
	if err := e.Reload(); err != nil {
		t.Fatalf("Reload (wire fleet channel auth): %v", err)
	}
	// The production watcher, not a hand-driven Reload: every convergence
	// bound below is therefore measuring the real 2s fingerprint poll.
	stopWatch := make(chan struct{})
	t.Cleanup(func() { close(stopWatch) })
	go e.Watch(stopWatch)

	// --- the fleet: two boxes on real reverse channels ------------------------
	stopA := startFleetBox(t, srv.URL, e2eFleet+"/box-a", tokenA, namedFleetHandler("box-a"))
	t.Cleanup(stopA)
	stopB := startFleetBox(t, srv.URL, e2eFleet+"/box-b", tokenB, namedFleetHandler("box-b"))
	t.Cleanup(stopB)
	waitChannel(t, hub, e2eFleet+"/box-a", "connected", e2eDialBound)
	waitChannel(t, hub, e2eFleet+"/box-b", "connected", e2eDialBound)

	// The platform's deployment bucket: where fleet boxes publish bindings and
	// where the platform box reads them back. One store, both roles — the
	// object IS the only channel a box has to say who is serving what.
	store := release.NewDirStore(t.TempDir())

	// box-b is the current holder: it owns the host's lease and has published
	// the takeover binding its last wake produced.
	leases := newFakeFleetLease(e2eLeaseTTL)
	if err := leases.acquire(e2eAppHost, "box-b"); err != nil {
		t.Fatalf("box-b initial lease: %v", err)
	}
	stopRenewB := leases.startRenewer(e2eAppHost, "box-b")
	t.Cleanup(stopRenewB)
	if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{
		Host: e2eAppHost, Fleet: e2eFleet, Box: "box-b", BoundAt: time.Now().UnixMilli(),
	}, true); err != nil {
		t.Fatalf("box-b binding: %v", err)
	}

	// --- (a) a bound route rides its bound box's channel ----------------------
	if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-b" {
		t.Fatalf("(a) got %d %q, want 200 %q — the BOUND box, which sorts last; a fallback-only edge would answer box-a", code, body, "box-b")
	}
	t.Log("(a) bound route rode the bound box's channel")

	// --- (b) the bound box goes dark ------------------------------------------
	// Its agent stops (power, Wi-Fi, a reboot) — so it renews nothing and
	// answers nothing.
	stopRenewB()
	stopB()

	// The lease is what stops a split brain, and it does not lapse the instant
	// the box dies: the survivor's very next wake attempt must LOSE while
	// box-b's hold is still live, with the typed error the activator keys off.
	err = leases.acquire(e2eAppHost, "box-a")
	if err == nil {
		t.Fatal("(b) box-a claimed the lease while box-b's hold was still live — that is a split brain, two boxes serving one host")
	}
	var held *lease.HeldError
	if !errors.As(err, &held) {
		t.Fatalf("(b) contested acquire returned %T (%v), want a *lease.HeldError", err, err)
	}

	// T4 CARRY (observation, not a gate): the edge does not learn a channel
	// died from the channel dying. Nothing unregisters a dead conn except
	// pingLoop — for a fleet channel that is fleetPingEvery (5s) +
	// fleetPingTimeout (5s), so up to ~10s of black hole (spec D7 narrowed it
	// from the static channels' ~25s) in which the hub still reports the box
	// connected and rule 1 keeps dispatching to a corpse instead of falling
	// over to the healthy survivor. That window IS the venue-Wi-Fi failure
	// mode: an app stays down for up to ~10s after its box drops off the
	// network, even though another box could have served it immediately.
	// Observed here without paying the 10s: the assertions below are that the
	// channel is STILL registered and the request does NOT reach box-a.
	if got := hub.State(e2eFleet + "/box-b"); got != "connected" {
		t.Fatalf("(b) hub state for the dead box = %q, want %q — if the hub now reaps eagerly, the black-hole note above is stale and this test should be updated to assert the better behavior", got, "connected")
	}
	resp, err := e2eDo(srv.URL, e2eAppHost, time.Second)
	if err == nil {
		defer resp.Body.Close()
		body, _ := io.ReadAll(resp.Body)
		if resp.StatusCode == http.StatusOK && string(body) == "box-a" {
			t.Fatal("(b) the request fell over to box-a before the dead channel was reaped — good news, but it means the ~10s black-hole window this test documents is gone and the comment above needs rewriting")
		}
		t.Logf("(b) black-hole window: dead-but-registered channel answered %d %q (no fallover)", resp.StatusCode, string(body))
	} else {
		t.Logf("(b) black-hole window: dead-but-registered channel hung until the client gave up (%v)", err)
	}

	// pingLoop's reap, on demand — CloseChannel is exactly what it calls, so
	// the rest of the sequence runs at the speed the fleet would see once the
	// ping has fired, not 10s later.
	hub.CloseChannel(e2eFleet + "/box-b")
	waitChannel(t, hub, e2eFleet+"/box-b", "", time.Second)

	// Rule 2: bound box dark, so any connected box of the fleet serves. The
	// app is up again without any registry write at all.
	if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" {
		t.Fatalf("(b) after the bound box went dark: got %d %q, want 200 %q (the fallback rule)", code, body, "box-a")
	}
	t.Log("(b) fell over to the survivor while still bound to the dead box")

	// The survivor wakes the app locally: it can only claim the lease once
	// box-b's hold LAPSES (nothing renews it — the box is dark), and only that
	// win authorizes the takeover binding.
	acquireDeadline := time.Now().Add(e2eLeaseTTL + 3*time.Second)
	var won bool
	for time.Now().Before(acquireDeadline) {
		if err := leases.acquire(e2eAppHost, "box-a"); err == nil {
			won = true
			break
		}
		time.Sleep(50 * time.Millisecond)
	}
	if !won {
		t.Fatalf("(b) box-a never won the lease after box-b's hold should have lapsed (TTL %s)", e2eLeaseTTL)
	}
	stopRenewA := leases.startRenewer(e2eAppHost, "box-a")
	t.Cleanup(stopRenewA)

	// The wake-success publish (fleetBindingHook's shape: takeover=true,
	// millisecond BoundAt) followed by the platform's reconcile. In production
	// the reconcile rides the agent's instancesLoop; calling it directly is
	// what makes this test deterministic instead of loop-timed.
	if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{
		Host: e2eAppHost, Fleet: e2eFleet, Box: "box-a", BoundAt: time.Now().UnixMilli(),
	}, true); err != nil {
		t.Fatalf("(b) box-a takeover binding: %v", err)
	}
	if err := host.ReconcileFleetBindings(ctx, store, reg); err != nil {
		t.Fatalf("(b) ReconcileFleetBindings: %v", err)
	}
	rt, err := reg.Get(e2eAppHost)
	if err != nil {
		t.Fatalf("(b) read route: %v", err)
	}
	if rt.FleetBox != "box-a" {
		t.Fatalf("(b) Route.FleetBox = %q, want box-a — the takeover binding never made it back into routing", rt.FleetBox)
	}
	t.Log("(b) survivor won the lapsed lease, published its takeover binding, and the reconcile restamped FleetBox")

	// --- (c) the dead box returns and loses the reclaim -----------------------
	stopB2 := startFleetBox(t, srv.URL, e2eFleet+"/box-b", tokenB, namedFleetHandler("box-b"))
	t.Cleanup(stopB2)
	waitChannel(t, hub, e2eFleet+"/box-b", "connected", e2eDialBound)

	// Its activator tries to resume serving the host it used to own. box-a's
	// renewer is keeping that lease alive, so the revenant loses — this is the
	// HeldError path, the one thing standing between "box-b rebooted" and "two
	// boxes writing one SQLite database".
	err = leases.acquire(e2eAppHost, "box-b")
	if err == nil {
		t.Fatal("(c) the returning box reclaimed a lease the survivor is actively renewing — split brain")
	}
	if !errors.As(err, &held) {
		t.Fatalf("(c) contested reclaim returned %T (%v), want a *lease.HeldError", err, err)
	}
	// No wake success means no takeover write. What box-b's agent DOES run on
	// startup is the non-stealing resident pass (publishResidentFleetBindings'
	// shape: takeover=false) over the route still sitting in its local
	// registry — and that must not claim the binding back.
	if err := host.WriteFleetBinding(ctx, store, host.FleetBinding{
		Host: e2eAppHost, Fleet: e2eFleet, Box: "box-b", BoundAt: time.Now().UnixMilli(),
	}, false); err != nil {
		t.Fatalf("(c) box-b startup pass: %v", err)
	}
	if err := host.ReconcileFleetBindings(ctx, store, reg); err != nil {
		t.Fatalf("(c) ReconcileFleetBindings: %v", err)
	}
	rt, err = reg.Get(e2eAppHost)
	if err != nil {
		t.Fatalf("(c) read route: %v", err)
	}
	if rt.FleetBox != "box-a" {
		t.Fatalf("(c) Route.FleetBox = %q, want box-a — the returning box stole routing back without holding the lease", rt.FleetBox)
	}
	// And end to end, with BOTH boxes connected: the survivor serves. Anything
	// that went wrong above — a restamp that never happened, a binding stolen
	// back — shows up here as "box-b", because rule 1 would point at box-b in
	// either case. Polled, because the restamp only reaches the request path
	// through the edge's 2s fingerprint poll.
	waitBody(t, srv.URL, e2eAppHost, "box-a", e2eConvergeBound)
	t.Log("(c) the returning box lost the reclaim; routing stayed on the survivor with both boxes connected")

	// --- (d) rotation is asymmetric, and converges within one poll ------------
	// Baseline first, so the probe's two answers are known to be
	// distinguishable BEFORE the rotation: box-a's current token authorizes
	// (400 = authorized, then refused for the missing Upgrade headers).
	if code := registerProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", tokenA); code != http.StatusBadRequest {
		t.Fatalf("(d) probe baseline = %d, want 400 (credential accepted, wire contract refused)", code)
	}
	const rotatedA = "tok-a2"
	if err := reg.SetFleetBoxToken(e2eFleet, "box-a", e2eHashToken(rotatedA)); err != nil {
		t.Fatalf("(d) rotate box-a: %v", err)
	}
	// Half one: a NEW dial with the OLD token is refused, within one poll.
	took := waitProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", tokenA, http.StatusUnauthorized, e2eConvergeBound)
	t.Logf("(d) rotation reached the edge's auth table in %s (bound %s)", took, e2eConvergeBound)
	// ...and the NEW token is what authorizes now — so the 401 above is a
	// rotation, not the channel having vanished from auth altogether (which is
	// what a detach does, and which would close the live session below).
	if code := registerProbe(t, srv.URL, u.Host, e2eFleet+"/box-a", rotatedA); code != http.StatusBadRequest {
		t.Fatalf("(d) probe with the rotated token = %d, want 400 (the new credential must authorize)", code)
	}
	// Half two, the asymmetry that is the whole design decision: the ALREADY
	// LIVE session — still running on the old token, because box-a's agent has
	// not been told its credential changed — keeps carrying traffic. If
	// rotation closed it, box-a's loop would re-dial with the stale token,
	// fail 401 forever, and this request would be answered by box-b instead.
	if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" {
		t.Fatalf("(d) after rotating box-a's token: got %d %q, want 200 %q — rotation must refuse new dials, never evict a live session", code, body, "box-a")
	}
	t.Log("(d) rotation refused the old token's next dial and left the live session serving")

	// Detach is the OTHER half of that asymmetry: it revokes now. box-b (idle,
	// connected, holding nothing) is taken out of the fleet, and its channel
	// must be closed by the same Watch->Reload->SetDynamicAuth path.
	if err := reg.SetFleetBoxStatus(e2eFleet, "box-b", "detached"); err != nil {
		t.Fatalf("(d) detach box-b: %v", err)
	}
	waitChannel(t, hub, e2eFleet+"/box-b", "", e2eConvergeBound)
	// And it STAYS down: its agent re-dials every few hundred ms and is
	// refused every time, because a detached box is gone from the auth table
	// entirely. A one-shot close that let the box straight back in would pass
	// the line above and fail here. 20x100ms (2s, M2 final-review widening
	// from 10x100ms/1s) gives a slow CI runner's re-dial loop more than one
	// chance to prove it stays refused, not just that it hasn't reconnected
	// yet.
	for i := 0; i < 20; i++ {
		if got := hub.State(e2eFleet + "/box-b"); got != "" {
			t.Fatalf("(d) detached box reconnected (state %q) — detach must revoke the credential, not just close the socket", got)
		}
		time.Sleep(100 * time.Millisecond)
	}
	// Detaching one box did not disturb the other.
	if code, body := e2eGet(t, srv.URL, e2eAppHost); code != http.StatusOK || body != "box-a" {
		t.Fatalf("(d) after detaching box-b: got %d %q, want 200 %q", code, body, "box-a")
	}
	t.Log("(d) detach closed the channel and kept it closed")

	// --- (e) both boxes dark: 503 + Retry-After -------------------------------
	stopA()
	hub.CloseChannel(e2eFleet + "/box-a") // pingLoop's reap again
	waitChannel(t, hub, e2eFleet+"/box-a", "", time.Second)
	if names := hub.ConnectedOfFleet(e2eFleet); len(names) != 0 {
		t.Fatalf("(e) fleet still has connected channels %v — the dark case is not set up", names)
	}
	dark, err := e2eDo(srv.URL, e2eAppHost, 10*time.Second)
	if err != nil {
		t.Fatalf("(e) GET against a dark fleet: %v", err)
	}
	defer dark.Body.Close()
	if dark.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("(e) dark fleet answered %d, want 503 (not a 502 that reads like a broken backend)", dark.StatusCode)
	}
	if got := dark.Header.Get("Retry-After"); got != "2" {
		t.Fatalf("(e) Retry-After = %q, want %q", got, "2")
	}
	t.Log("(e) dark fleet answered 503 with Retry-After: 2")
}