// Task 5 of the fleet-api plan: the spec's central promise proved end to end
// — a fleet box holding NO bucket credentials converges entirely through the
// store door (spec 2026-08-11-fleet-api-surface-design.md, F1-F8; issue #73).
//
// The box's whole view of the platform's deployment bucket is a
// fleetapi.New(...) client pointed at a REAL edge: real TunnelHub auth
// (registry-backed, the same table detach/rotate converge), the real
// fleet→account snapshot Reload builds, the real ACL, a DirStore behind it.
// No box-side code path in this file ever touches that DirStore — every
// platform-side use of it below is the OTHER role (the console/CLI writing
// the bucket, or the assertion reading back what a box wrote through the
// door).
//
// It sits beside fleet_e2e_test.go rather than inside it because the two
// prove different things about the same fleet and want different worlds:
// TestTwoBoxFleetEndToEnd drives the ROUTING failure sequence (channels,
// leases, takeover, dark fleet) and its edge deliberately has no static
// store — no door at all. This one drives the STORE loop (adopt, serial,
// provision, heartbeat, bindings, logs) and needs EnableStatic wired. What
// they do share is shared: e2eHashToken, startFleetBox, namedFleetHandler,
// waitChannel, e2eGet and waitBody all come from that file, so the edge/hub
// arrangement has exactly one source of truth per concern.
//
// Hermetic and fast: loopback httptest edge, real h2 reverse channels via
// the production tunnelclient.Loop, real ed25519 release signing.
package edge_test

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"net/http/httptest"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/applog"
	"github.com/carlosframework/platform/internal/domains"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/fleetapi"
	"github.com/carlosframework/platform/internal/host"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

const (
	fapiFleet  = "orchard"
	fapiBoxA   = "box-a"
	fapiBoxB   = "box-b"
	fapiTokenA = "orchard-token-a"
	fapiTokenB = "orchard-token-b"

	fapiApp     = "shop"
	fapiDomain  = "orchard.example"
	fapiHost    = "acme." + fapiDomain
	fapiRegion  = "ie"
	fapiChannel = "canary/e2e" // canary/* has no bake window: the pass applies at once
	fapiVersion = "v1"

	// The two account identities. Both are real sqids (domains.Sqid of an id),
	// never literals: provision.knownAccounts refuses any account manifest
	// whose path sqid, body sqid and Sqid(body.ID) do not all agree, so a
	// hardcoded string would quietly stop being an account the moment the
	// frozen encoding changed — and this test's whole provisioning leg would
	// then pass for the wrong reason (nothing known, nothing minted).
	fapiAcctID   int64 = 7
	fapiOtherID  int64 = 8
	fapiOtherApp       = "intruder"
)

// fapiPlatform is the artifact key a box of THIS machine's shape adopts —
// release's own adoptPlatform, which is unexported.
var fapiPlatform = runtime.GOOS + "-" + runtime.GOARCH

// fleetDoorWorld is one platform + one fleet: the edge and its bucket on one
// side, two fleet boxes whose only bucket access is the door on the other.
type fleetDoorWorld struct {
	reg    *registry.Registry // the PLATFORM's registry: fleet rows, box credentials, fleet-placed routes
	store  *release.DirStore  // the platform's deployment bucket — edge-side and console-side handle
	srvURL string
	hub    *edge.TunnelHub
	priv   ed25519.PrivateKey
	pub    ed25519.PublicKey
	sqid   string // the fleet's owning account
	other  string // a second tenant on the same bucket, which the fleet must never see
	boxA   release.Store
	boxB   release.Store
}

// newFleetDoorWorld stands the world up and returns it with both boxes'
// channels connected — the same credential that dials the tunnel is the one
// their store clients authenticate with, which is F1/F2 stated as a fixture.
func newFleetDoorWorld(t *testing.T) *fleetDoorWorld {
	t.Helper()
	w := &fleetDoorWorld{
		sqid:  domains.Sqid(fapiAcctID),
		other: domains.Sqid(fapiOtherID),
	}

	// Not closed at cleanup, for the reason fleet_e2e_test.go gives: Edge.Watch
	// may still be mid-poll when the test returns.
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	w.reg = reg
	if err := reg.UpsertFleet(registry.Fleet{Name: fapiFleet, Account: w.sqid, Bucket: "orchard-data"}); err != nil {
		t.Fatalf("UpsertFleet: %v", err)
	}
	for label, tok := range map[string]string{fapiBoxA: fapiTokenA, fapiBoxB: fapiTokenB} {
		if err := reg.AddFleetBox(registry.FleetBox{
			Fleet: fapiFleet, Label: label, TokenHash: e2eHashToken(tok), Status: "active",
		}); err != nil {
			t.Fatalf("AddFleetBox(%s): %v", label, err)
		}
	}

	w.store = release.NewDirStore(t.TempDir())
	seedB64, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatalf("GenerateKey: %v", err)
	}
	if w.priv, err = release.ParseSigningKey(seedB64, "test"); err != nil {
		t.Fatalf("ParseSigningKey: %v", err)
	}
	if w.pub, err = release.ParsePublicKey(pubB64, "test"); err != nil {
		t.Fatalf("ParsePublicKey: %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)
	w.srvURL = srv.URL
	u, err := url.Parse(srv.URL)
	if err != nil {
		t.Fatal(err)
	}
	// EnableStatic is what hands the door its store handle — the edge's OWN,
	// never a second client, and the door is only reachable on the hub's
	// reserved hostname (here the listener's own, as in production
	// CARLOS_TUNNEL_HOST).
	w.hub = edge.NewTunnelHub(u.Hostname(), nil)
	e.EnableStatic(w.store, w.pub, filepath.Join(t.TempDir(), "cache"))
	e.EnableTunnels(w.hub)
	if err := e.Reload(); err != nil {
		t.Fatalf("Reload: %v", err)
	}
	stopWatch := make(chan struct{})
	t.Cleanup(func() { close(stopWatch) })
	go e.Watch(stopWatch)

	stopA := startFleetBox(t, srv.URL, fapiFleet+"/"+fapiBoxA, fapiTokenA, namedFleetHandler(fapiBoxA))
	t.Cleanup(stopA)
	stopB := startFleetBox(t, srv.URL, fapiFleet+"/"+fapiBoxB, fapiTokenB, namedFleetHandler(fapiBoxB))
	t.Cleanup(stopB)
	waitChannel(t, w.hub, fapiFleet+"/"+fapiBoxA, "connected", e2eDialBound)
	waitChannel(t, w.hub, fapiFleet+"/"+fapiBoxB, "connected", e2eDialBound)

	// The boxes' deployment stores. srv.URL is the door's base exactly as
	// CARLOS_TUNNEL_URL is in production, and the name is the box's own
	// channel identity — no AWS config, no bucket name, no credentials.
	w.boxA = fapiStore(t, srv.URL, fapiBoxA, fapiTokenA)
	w.boxB = fapiStore(t, srv.URL, fapiBoxB, fapiTokenB)
	return w
}

func fapiStore(t *testing.T, base, label, token string) release.Store {
	t.Helper()
	s, err := fleetapi.New(base, fapiFleet+"/"+label, token)
	if err != nil {
		t.Fatalf("fleetapi.New(%s): %v", label, err)
	}
	return s
}

// put is a platform-side (console/CLI-shaped) write straight to the bucket:
// create-only, the same shape every real writer of these keys uses.
func (w *fleetDoorWorld) put(t *testing.T, key, body string) {
	t.Helper()
	if _, err := w.store.Put(context.Background(), key, []byte(body), ""); err != nil {
		t.Fatalf("seed %s: %v", key, err)
	}
}

// accountManifest is domains' account record at the key provision reads —
// accounts/<sqid>/account.json, one level deep (never accounts/<sqid>.json).
func fapiAccountManifest(id int64, sqid, name string) string {
	b, _ := json.Marshal(map[string]any{
		"id": id, "sqid": sqid, "name": name, "created_at": "2026-08-11T09:00:00Z",
	})
	return string(b)
}

// TestFleetBoxConvergesWithoutBucketCredentials is Task 5's acceptance
// evidence. The legs run in sequence against ONE world, each depending on
// what the last left behind (a record becomes a route, the route adopts a
// release, the box heartbeats about it, its binding lands in routing) — a
// failure means the loop stopped there and nothing after it is meaningful.
func TestFleetBoxConvergesWithoutBucketCredentials(t *testing.T) {
	ctx := context.Background()
	w := newFleetDoorWorld(t)

	// A second tenant on the same bucket, planted BEFORE anything else so
	// every listing below runs against a store that really does hold another
	// account's keys — the negative leg (f) is only meaningful if the data it
	// says is invisible is genuinely there.
	w.put(t, "accounts/"+w.other+"/account.json", fapiAccountManifest(fapiOtherID, w.other, "Intruder"))
	otherChannelKey := "apps/" + w.other + "/" + fapiOtherApp + "/channels/stable.json"
	w.put(t, otherChannelKey, `{"app":"intruder","version":"v9"}`)
	otherLogKey, err := applog.ChunkKey(w.other, fapiOtherApp, fapiBoxA, applog.HourOf(time.Now()), 1, 9)
	if err != nil {
		t.Fatal(err)
	}
	w.put(t, otherLogKey, "not yours")
	w.put(t, "fleets/other/status/boxes/box-a.json", `{"label":"box-a"}`)

	// --- (a) the 2s convergence poll: control/serial, read-only --------------
	// The one shared platform key a box polls every 2 seconds. It must read,
	// and it must never be writable — a box that could bump the serial could
	// make every other box in the deployment run a full pass set on demand.
	w.put(t, release.SerialKey, "41")
	serial, _, err := w.boxA.Get(ctx, release.SerialKey)
	if err != nil || string(serial) != "41" {
		t.Fatalf("(a) serial through the door = %q, %v; want \"41\"", serial, err)
	}
	_, etag, err := w.store.Get(ctx, release.SerialKey)
	if err != nil {
		t.Fatalf("(a) read serial etag: %v", err)
	}
	if _, err := w.store.Put(ctx, release.SerialKey, []byte("42"), etag); err != nil {
		t.Fatalf("(a) console bump: %v", err)
	}
	if serial, _, err = w.boxA.Get(ctx, release.SerialKey); err != nil || string(serial) != "42" {
		t.Fatalf("(a) serial after the console bump = %q, %v; want \"42\" — the poll would never see a deploy", serial, err)
	}
	if _, err := w.boxA.Put(ctx, release.SerialKey, []byte("999"), ""); err == nil {
		t.Fatal("(a) a fleet box bumped control/serial through the door")
	}
	t.Log("(a) control/serial reads through the door and refuses the box's write")

	// --- (b) provisioning: an instance record becomes a local route ----------
	// The account-scoped LISTING path end to end: knownAccounts lists
	// accounts/ (a shared prefix, filtered per tenant), the record scan lists
	// apps/ (scope-mapped to this account's subtree), and the pass mints.
	w.put(t, "accounts/"+w.sqid+"/account.json", fapiAccountManifest(fapiAcctID, w.sqid, "Orchard"))
	if err := provision.PutManifest(ctx, w.store, w.sqid, fapiApp, provision.Manifest{
		Schema: 1, App: fapiApp, Domain: fapiDomain, Regions: []string{fapiRegion}, Seed: []string{"serve"},
	}); err != nil {
		t.Fatalf("(b) PutManifest: %v", err)
	}
	if err := provision.CreateRecord(ctx, w.store, provision.Record{
		Schema: 1, Host: fapiHost, Slug: "acme", App: fapiApp, Account: w.sqid,
		Region: fapiRegion, Channel: fapiChannel, CreatedAt: time.Now(),
	}); err != nil {
		t.Fatalf("(b) CreateRecord: %v", err)
	}
	// The other tenant's record, in THIS box's region and pool: the only thing
	// keeping it off this box is the door's account scope.
	if err := provision.CreateRecord(ctx, w.store, provision.Record{
		Schema: 1, Host: "evil." + fapiDomain, Slug: "evil", App: fapiOtherApp, Account: w.other,
		Region: fapiRegion, Channel: "stable", CreatedAt: time.Now(),
	}); err != nil {
		t.Fatalf("(b) CreateRecord (other tenant): %v", err)
	}

	// The box's OWN registry — a fleet box has one, separate from the
	// platform's, and this pass is what puts routes in it.
	boxReg, err := registry.Open(filepath.Join(t.TempDir(), "box-registry.db"))
	if err != nil {
		t.Fatalf("(b) open box registry: %v", err)
	}
	dataDir := t.TempDir()
	if err := provision.Reconcile(ctx, w.boxA, boxReg, provision.ReconcileConfig{
		Region: fapiRegion, DataDir: dataDir,
	}); err != nil {
		t.Fatalf("(b) provision.Reconcile through the door: %v", err)
	}
	minted, err := boxReg.Get(fapiHost)
	if err != nil || minted == nil {
		t.Fatalf("(b) route %s = %+v, %v — the record never became a route", fapiHost, minted, err)
	}
	if minted.App != fapiApp || minted.Account != w.sqid || minted.Channel != fapiChannel ||
		!minted.Provisioned || minted.Seed != `["serve"]` {
		t.Fatalf("(b) minted route = %+v, want the record's app/account/channel, Provisioned and the manifest seed", minted)
	}
	if rt, err := boxReg.Get("evil." + fapiDomain); err != nil || rt != nil {
		t.Fatalf("(b) the other tenant's record minted a route on this box: %+v, %v", rt, err)
	}
	t.Log("(b) provision.Reconcile minted the fleet's own instance route and could not see the other tenant's")

	// --- (c) release adoption: pointer, artifact, staging --------------------
	// The whole adoption pass over the door: the signed channel pointer is
	// fetched and verified, the manifest read, the artifact BODY streamed
	// through the edge (F8 — no presigned redirect), and the version staged
	// and stamped on the route.
	artifact := make([]byte, 64<<10)
	for i := range artifact {
		artifact[i] = byte(i * 7)
	}
	if err := release.Ship(ctx, w.store, w.sqid, release.Manifest{
		App: fapiApp, Version: fapiVersion, Kind: "binary",
		Artifacts: map[string]release.Artifact{fapiPlatform: {}},
	}, map[string][]byte{fapiPlatform: artifact}); err != nil {
		t.Fatalf("(c) Ship: %v", err)
	}
	if err := release.Promote(ctx, w.store, w.priv, w.sqid, fapiApp, fapiChannel, fapiVersion); err != nil {
		t.Fatalf("(c) Promote: %v", err)
	}
	optDir := t.TempDir()
	var restarted []string
	restart := func(h string) error { restarted = append(restarted, h); return nil }
	if err := release.Pass(ctx, boxReg, w.boxA, w.pub, optDir, restart); err != nil {
		t.Fatalf("(c) release.Pass through the door: %v", err)
	}
	if v, err := release.LiveVersion(optDir, fapiHost); v != fapiVersion {
		t.Fatalf("(c) live version = %q (%v), want %q — adoption did not converge through the door", v, err, fapiVersion)
	}
	staged, err := os.ReadFile(release.VersionPath(optDir, w.sqid, fapiVersion, fapiApp))
	if err != nil || !slices.Equal(staged, artifact) {
		t.Fatalf("(c) staged artifact = %d bytes (%v), want the shipped %d — a truncated body would pass a shorter check", len(staged), err, len(artifact))
	}
	adopted, err := boxReg.Get(fapiHost)
	if err != nil || adopted == nil || adopted.Version != fapiVersion {
		t.Fatalf("(c) route version = %+v, %v; want %q stamped", adopted, err, fapiVersion)
	}
	t.Logf("(c) adoption converged: pointer verified, %d artifact bytes streamed and staged (restarted %v)", len(artifact), restarted)

	// --- (d) the heartbeat, under the Task-4 default prefix ------------------
	// A door-backed box gets CARLOS_STATUS_PREFIX defaulted to
	// fleets/<fleet>/ (spec F5) precisely because the shared status/ root is
	// outside the door's write ACL. The two lines statusLoop runs, with that
	// rebase and the agent's label override:
	prefixed := release.NewPrefixStore(w.boxA, "fleets/"+fapiFleet+"/")
	routes, err := boxReg.List()
	if err != nil {
		t.Fatalf("(d) list box routes: %v", err)
	}
	box := status.CollectBox(routes, status.Probes{MachineID: "orchard-a", Region: "IE"}, time.Now())
	box.Label = fapiBoxA // statusTick's labelOverride: the fleet label, not a machine-id hash
	if err := status.WriteBox(ctx, prefixed, box); err != nil {
		t.Fatalf("(d) status.WriteBox through the door: %v", err)
	}
	heartbeatKey := "fleets/" + fapiFleet + "/status/boxes/" + fapiBoxA + ".json"
	raw, _, err := w.store.Get(ctx, heartbeatKey)
	if err != nil {
		t.Fatalf("(d) heartbeat did not land at %s: %v", heartbeatKey, err)
	}
	var landed status.Box
	if err := json.Unmarshal(raw, &landed); err != nil {
		t.Fatalf("(d) heartbeat body: %v", err)
	}
	if landed.Label != fapiBoxA {
		t.Fatalf("(d) heartbeat label = %q, want %q", landed.Label, fapiBoxA)
	}
	// The SECOND write is the one that exercises CAS through the door:
	// WriteBox reads its own object's etag first (the read half of the fleet
	// corner, which only exists so this replace can happen) and Puts with
	// If-Match. A door that dropped etags would fail here, not above.
	if err := status.WriteBox(ctx, prefixed, box); err != nil {
		t.Fatalf("(d) heartbeat replace (read-etag-then-Put) through the door: %v", err)
	}
	// And a box may not heartbeat into a fleet that is not its own.
	if _, err := w.boxA.Put(ctx, "fleets/other/status/boxes/box-a.json", []byte("{}"), ""); err == nil {
		t.Fatal("(d) box-a wrote into another fleet's corner")
	}
	t.Logf("(d) heartbeat landed at %s and replaced itself under CAS", heartbeatKey)

	// --- (e) a fleet binding takeover, box → bucket → routing ----------------
	// The platform's route for the app, placed on the fleet and currently
	// bound to box-b.
	if err := w.reg.Upsert(registry.Route{
		Host: fapiHost, Kind: "instance", Status: "ready",
		Region: registry.PlacementTarget(fapiFleet), FleetBox: fapiBoxB,
	}); err != nil {
		t.Fatalf("(e) upsert fleet-placed route: %v", err)
	}
	// box-b's agent-startup resident pass (takeover=false), through ITS door
	// client — an observation, with no lease behind it.
	if err := host.WriteFleetBinding(ctx, w.boxB, host.FleetBinding{
		Host: fapiHost, Fleet: fapiFleet, Box: fapiBoxB, BoundAt: time.Now().UnixMilli(),
	}, false); err != nil {
		t.Fatalf("(e) box-b resident binding: %v", err)
	}
	// box-a wakes the app and wins the lease (modeled in fleet_e2e_test.go;
	// what matters here is the WRITE): a takeover claims the binding.
	if err := host.WriteFleetBinding(ctx, w.boxA, host.FleetBinding{
		Host: fapiHost, Fleet: fapiFleet, Box: fapiBoxA, BoundAt: time.Now().UnixMilli(),
	}, true); err != nil {
		t.Fatalf("(e) box-a takeover binding through the door: %v", err)
	}
	// The platform reads it back from its own bucket handle — the object IS
	// the message — and restamps routing.
	if err := host.ReconcileFleetBindings(ctx, w.store, w.reg); err != nil {
		t.Fatalf("(e) ReconcileFleetBindings: %v", err)
	}
	rt, err := w.reg.Get(fapiHost)
	if err != nil || rt == nil || rt.FleetBox != fapiBoxA {
		t.Fatalf("(e) Route.FleetBox = %+v, %v; want %q — the takeover never made it back into routing", rt, err, fapiBoxA)
	}
	// And the guard survives the round trip: box-b's next resident pass reads
	// the existing binding through the door and declines to steal it back.
	if err := host.WriteFleetBinding(ctx, w.boxB, host.FleetBinding{
		Host: fapiHost, Fleet: fapiFleet, Box: fapiBoxB, BoundAt: time.Now().UnixMilli(),
	}, false); err != nil {
		t.Fatalf("(e) box-b second resident pass: %v", err)
	}
	if err := host.ReconcileFleetBindings(ctx, w.store, w.reg); err != nil {
		t.Fatalf("(e) ReconcileFleetBindings (after box-b's pass): %v", err)
	}
	if rt, err = w.reg.Get(fapiHost); err != nil || rt == nil || rt.FleetBox != fapiBoxA {
		t.Fatalf("(e) Route.FleetBox = %+v, %v; want %q — a non-takeover write stole the binding through the door", rt, err, fapiBoxA)
	}
	// End to end, from outside: the edge's 2s poll picks the restamp up and
	// the app is served by box-a's own channel. Record → route → release →
	// binding → traffic, with the box holding no bucket credential anywhere
	// in that chain.
	waitBody(t, w.srvURL, fapiHost, fapiBoxA, e2eConvergeBound)
	t.Log("(e) the takeover written through the door reached routing, and the bound box serves")

	// --- (f) the other tenant is invisible ----------------------------------
	// Listings first: the easiest place to leak a neighbour, and the one
	// provision.Reconcile above depends on being right.
	appKeys, err := w.boxA.List(ctx, "apps/")
	if err != nil {
		t.Fatalf("(f) list apps/: %v", err)
	}
	for _, k := range appKeys {
		if !strings.HasPrefix(k, "apps/"+w.sqid+"/") {
			t.Fatalf("(f) list apps/ returned %q — outside this fleet's account", k)
		}
	}
	if len(appKeys) == 0 {
		t.Fatal("(f) list apps/ came back empty — the scope map is refusing the fleet's OWN estate")
	}
	acctKeys, err := w.boxA.List(ctx, "accounts/")
	if err != nil {
		t.Fatalf("(f) list accounts/: %v", err)
	}
	if !slices.Equal(acctKeys, []string{"accounts/" + w.sqid + "/account.json"}) {
		t.Fatalf("(f) list accounts/ = %v, want only this fleet's own account record", acctKeys)
	}
	acctDirs, err := w.boxA.ListDirs(ctx, "accounts/")
	if err != nil {
		t.Fatalf("(f) listdirs accounts/: %v", err)
	}
	if !slices.Equal(acctDirs, []string{"accounts/" + w.sqid + "/"}) {
		t.Fatalf("(f) listdirs accounts/ = %v, want only this fleet's own directory (one child per account exists)", acctDirs)
	}
	fleetDirs, err := w.boxA.ListDirs(ctx, "fleets/")
	if err != nil {
		t.Fatalf("(f) listdirs fleets/: %v", err)
	}
	for _, d := range fleetDirs {
		if !strings.HasPrefix(d, "fleets/"+fapiFleet+"/") {
			t.Fatalf("(f) listdirs fleets/ returned %q — another fleet's corner", d)
		}
	}
	// Then direct gets: every one of these keys exists in the bucket, and
	// every one of them is os.ErrNotExist to this box.
	for _, key := range []string{
		"accounts/" + w.other + "/account.json",
		otherChannelKey,
		otherLogKey,
		"apps/" + w.other + "/" + fapiOtherApp + "/instances/evil." + fapiDomain + ".json",
		"fleets/other/status/boxes/box-a.json",
	} {
		if _, _, err := w.store.Get(ctx, key); err != nil {
			t.Fatalf("(f) fixture %s is not actually in the bucket: %v", key, err)
		}
		if _, _, err := w.boxA.Get(ctx, key); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("(f) GET %s through the door = %v, want os.ErrNotExist (byte-identical 404)", key, err)
		}
	}
	// Writes and deletes into the neighbour are refused too — including the
	// delete of a log chunk, which is the one write verb the account-wide logs
	// grant makes tempting.
	if _, err := w.boxA.Put(ctx, otherChannelKey, []byte(`{"version":"evil"}`), ""); err == nil {
		t.Fatal("(f) box-a promoted a release in another account")
	}
	if _, err := w.boxA.Put(ctx, otherLogKey, []byte("forged"), ""); err == nil {
		t.Fatal("(f) box-a forged a log chunk in another account")
	}
	if err := w.boxA.Delete(ctx, otherLogKey); err == nil {
		t.Fatal("(f) box-a deleted another account's log chunk")
	}
	if _, _, err := w.store.Get(ctx, otherChannelKey); err != nil {
		t.Fatalf("(f) the refused writes changed the neighbour's object anyway: %v", err)
	}
	t.Log("(f) the second tenant is invisible to this fleet: listings, gets, writes and deletes all refuse")

	// --- (g) log chunks: ship and prune through the door ---------------------
	// The Task-4 grant, exercised as logshipLoop uses it: chunks are written
	// UNPREFIXED under the app's own subtree (applog.Prefix — where `carlos
	// logs` reads them), listed back, and pruned by the same box.
	hour := applog.HourOf(time.Now())
	chunk, err := applog.EncodeChunk([]applog.Record{{
		TS: time.Now(), Box: fapiBoxA, Host: fapiHost, Stream: "stdout", Msg: "listening", Seq: 1,
	}})
	if err != nil {
		t.Fatalf("(g) EncodeChunk: %v", err)
	}
	chunkKey, err := applog.ChunkKey(w.sqid, fapiApp, fapiBoxA, hour, 1, len(chunk))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.boxA.Put(ctx, chunkKey, chunk, ""); err != nil {
		t.Fatalf("(g) ship a log chunk through the door: %v", err)
	}
	landedChunk, _, err := w.store.Get(ctx, chunkKey)
	if err != nil {
		t.Fatalf("(g) chunk did not land at %s: %v", chunkKey, err)
	}
	recs, err := applog.DecodeChunk(landedChunk)
	if err != nil || len(recs) != 1 || recs[0].Msg != "listening" {
		t.Fatalf("(g) chunk round trip = %+v, %v", recs, err)
	}
	boxPrefix, err := applog.BoxPrefix(w.sqid, fapiApp, fapiBoxA)
	if err != nil {
		t.Fatal(err)
	}
	listed, err := w.boxA.List(ctx, boxPrefix)
	if err != nil || !slices.Contains(listed, chunkKey) {
		t.Fatalf("(g) list own log subtree = %v, %v; want %s", listed, err, chunkKey)
	}
	if err := w.boxA.Delete(ctx, chunkKey); err != nil {
		t.Fatalf("(g) prune own chunk through the door: %v", err)
	}
	if _, _, err := w.store.Get(ctx, chunkKey); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("(g) chunk survived its prune: %v", err)
	}
	// The grant stays segment-exact: the same account, the same app, one
	// segment over, is still refused.
	for _, key := range []string{
		release.ChannelKey(w.sqid, fapiApp, fapiChannel),
		provision.RecordKey(w.sqid, fapiApp, fapiHost),
		"apps/" + w.sqid + "/" + fapiApp + "/config/production.json",
	} {
		if _, err := w.boxA.Put(ctx, key, []byte("{}"), ""); err == nil {
			t.Fatalf("(g) the logs grant let the box write %s", key)
		}
	}
	t.Log("(g) log chunks ship, list and prune through the door; neighbouring keys stay refused")
}

// TestFleetBoxCatchallYieldsToAnotherTenantsClaim (#215 review): the store
// door withholds another account's claim BODY from a fleet box, but must
// still list its NAME. An instance catch-all derives no <label>.<parent>
// alias over a name somebody else has claimed, and ReconcileAliases decides
// "somebody else has claimed it" from the listing; a listing narrowed to the
// box's own account let the box shadow the other tenant's claim.
func TestFleetBoxCatchallYieldsToAnotherTenantsClaim(t *testing.T) {
	w := newFleetDoorWorld(t)
	ctx := context.Background()
	const parent = "teams.example.com"
	if _, _, err := domains.AttachClaim(ctx, w.store, domains.Claim{Host: parent, Account: w.sqid, App: "jelly"}); err != nil {
		t.Fatal(err)
	}
	if _, err := domains.AttachInstanceCatchall(ctx, w.store, parent, w.sqid, "jelly", "", ""); err != nil {
		t.Fatal(err)
	}
	// The other tenant's exact claim on the name the catch-all would derive.
	if _, _, err := domains.AttachClaim(ctx, w.store, domains.Claim{
		Host: "beta." + parent, Account: w.other, App: "shop", Route: "shop." + w.other + ".oncarlos.com",
		CreatedBy: "sam@keymail.dev",
	}); err != nil {
		t.Fatal(err)
	}
	instance := "beta." + w.sqid + ".oncarlos.com"
	w.put(t, "apps/"+w.sqid+"/jelly/instances/"+instance+".json", `{}`)
	// A second instance whose name nobody else claims: the control that
	// shows the catch-all does derive through the door.
	control := "acme." + w.sqid + ".oncarlos.com"
	w.put(t, "apps/"+w.sqid+"/jelly/instances/"+control+".json", `{}`)

	// The box sees the other tenant's NAME but not its claim.
	keys, err := w.boxA.List(ctx, domains.ClaimsPrefix)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Contains(keys, domains.ClaimKey("beta."+parent)) {
		t.Fatalf("listing %v omits another tenant's claimed name", keys)
	}
	if _, _, err := w.boxA.Get(ctx, domains.ClaimKey("beta."+parent)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("another tenant's claim body: err = %v, want not-found", err)
	}

	reg, err := registry.Open(filepath.Join(t.TempDir(), "box.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: instance, Slug: "beta", App: "jelly", Account: w.sqid, Kind: "instance", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetProvisioned(instance, true); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: control, Slug: "acme", App: "jelly", Account: w.sqid, Kind: "instance", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetProvisioned(control, true); err != nil {
		t.Fatal(err)
	}
	if err := domains.ReconcileAliases(ctx, w.boxA, reg, time.Now, nil, nil); err != nil {
		t.Fatal(err)
	}
	aliases, err := reg.ListAliases()
	if err != nil {
		t.Fatal(err)
	}
	derived := false
	for _, a := range aliases {
		if a.Alias == "beta."+parent {
			t.Fatalf("the box derived %+v over another tenant's claim", a)
		}
		derived = derived || (a.Alias == "acme."+parent && a.Host == control)
	}
	if !derived {
		t.Fatalf("aliases = %+v, want acme.%s derived for the unclaimed name", aliases, parent)
	}
}

// Settings spec §4.1, ruling R17: a fleet box reads its own account's
// promotion records through the store door, so a build authorized by record
// alone (no pointer names it) stages on a fleet box, and a manifest that does
// not hash to the record's digest is refused. Another account's record and a
// listing of the records stay out of reach, and a box cannot write one.
func TestFleetBoxReadsItsOwnPromotionRecords(t *testing.T) {
	ctx := context.Background()
	w := newFleetDoorWorld(t)
	const version = "v7"
	for _, acct := range []string{w.sqid, w.other} {
		if err := release.Ship(ctx, w.store, acct, release.Manifest{
			App: fapiApp, Version: version, Kind: "binary",
			Artifacts: map[string]release.Artifact{fapiPlatform: {}},
		}, map[string][]byte{fapiPlatform: []byte("authorized bytes")}); err != nil {
			t.Fatalf("Ship(%s): %v", acct, err)
		}
		_, raw, err := release.GetManifestRaw(ctx, w.store, acct, fapiApp, version)
		if err != nil {
			t.Fatal(err)
		}
		if err := release.RootSigner(w.priv).WritePromotionRecord(ctx, w.store, release.PromotionRecord{
			Account: acct, App: fapiApp, Version: version, ManifestDigest: release.ManifestDigest(raw), Authorized: true,
		}); err != nil {
			t.Fatalf("WritePromotionRecord(%s): %v", acct, err)
		}
	}
	_, raw, err := release.GetManifestRaw(ctx, w.store, w.sqid, fapiApp, version)
	if err != nil {
		t.Fatal(err)
	}

	rec, err := release.ReadPromotionRecord(ctx, w.boxA, w.pub, w.sqid, fapiApp, version)
	if err != nil || rec.ManifestDigest != release.ManifestDigest(raw) {
		t.Fatalf("ReadPromotionRecord through the door = %+v, %v; want the record signing %s", rec, err, release.ManifestDigest(raw))
	}
	if m, err := release.AuthorizedManifest(ctx, w.boxA, w.pub, w.sqid, fapiApp, version); err != nil || m.Version != version {
		t.Fatalf("AuthorizedManifest through the door = %+v, %v; want the record-only authorized %s", m, err, version)
	}

	if _, err := release.ReadPromotionRecord(ctx, w.boxA, w.pub, w.other, fapiApp, version); err == nil {
		t.Fatal("a fleet box read another account's promotion record through the door")
	}
	keys, _ := w.boxA.List(ctx, "releases/"+w.sqid+"/")
	for _, k := range keys {
		if strings.HasPrefix(k, "releases/") {
			t.Fatalf("a fleet box listed promotion records through the door: %v", keys)
		}
	}
	if _, err := w.boxA.Put(ctx, release.PromotionRecordKey(w.sqid, fapiApp, "v8"), []byte("{}"), ""); err == nil {
		t.Fatal("a fleet box wrote a promotion record through the door")
	}

	_, etag, err := w.store.Get(ctx, release.ManifestKey(w.sqid, fapiApp, version))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.store.Put(ctx, release.ManifestKey(w.sqid, fapiApp, version), append(raw, '\n'), etag); err != nil {
		t.Fatal(err)
	}
	if _, err := release.AuthorizedManifest(ctx, w.boxA, w.pub, w.sqid, fapiApp, version); !errors.Is(err, release.ErrManifestDigest) {
		t.Fatalf("AuthorizedManifest over a replaced manifest = %v; want ErrManifestDigest", err)
	}
}